使用圖建置應用程式
當您只想載入並執行單一已編譯的模型封存檔時,請使用 Model。當您想要圍繞模型和節點建置應用程式時,請使用 Graph:新增公開的輸入和輸出,連接可重複使用的片段,分支資料流,合併資料流,驗證應用程式,並儲存或視覺化實際執行的內容。
其概念模型刻意保持簡潔:
| 概念 | 意義 |
|---|---|
Model | 從磁碟載入的已編譯模型封存檔,例如 resnet50.tar.gz 或 yolov8.tar.gz。 |
Node | 一個處理步驟:輸入、輸出、轉換、來源、接收器、模型階段或輔助階段。 |
Graph | 應用程式的連接計畫:存在哪些節點/片段,以及資料如何在它們之間流動。 |
Run | 由 Graph::build() 傳回的即時執行句柄:推送輸入、提取輸出、收集指標、停止。 |
簡而言之:
Graph = what to run
Run = the running instance
當您需要更短的路徑時,請從任務頁面開始:
大多數應用程式程式碼應使用公開的 simaai::neat::Graph 和 simaai::neat::Run。 請勿使用較低層級的實作命名空間來建立應用程式;這些不是客戶 API。
何時需要一個圖?
| 目標 | 建議的 API |
|---|---|
| 在單一輸入上執行一個模型 | Model::run(...) 或 Model::build(...) |
| 在模型周圍新增應用程式輸入/輸出邊界 | Graph |
| 使用自訂處理節點來組合一個模型 | Graph::add(...) |
| 在多個應用程式中重複使用 Graph 片段 | 傳回或傳遞一個 Graph 片段 |
| 路由多個輸入或輸出 | 命名 nodes::Input(...) / nodes::Output(...),再加上 connect(...) |
| 將一個資料流分支到多個消費者 | graphs::Branch(...) |
| 將多個資料流組合為一個邏輯輸出 | 使用 graphs::Combine(...) 和一個 CombinePolicy |
| 儲存或視覺化執行的拓 撲和指標 | save_run_json(run, ...) |
第一個圖:一個輸入、一個模型、一個輸出
這是最小的完整應用程式樣式圖:
#include <neat.h>
#include <iostream>
namespace neat = simaai::neat;
int main() {
neat::Model model("resnet50.tar.gz");
neat::Graph app;
app.add(neat::nodes::Input("image"));
app.add(model);
app.add(neat::nodes::Output("classes"));
neat::Run run = app.build();
neat::Tensor image = /* create or load an image tensor */;
run.push("image", neat::TensorList{image});
std::optional<neat::Sample> result = run.pull("classes", /*timeout_ms=*/1000);
if (result) {
// Consume result->tensors, result->detections, or other Sample metadata.
}
run.stop();
}
逐行說明:
nodes::Input("image")宣告一個名為image的公開輸入節點。app.add(model)將模型的選定路徑插入到圖中。nodes::Output("classes")宣告一個名為classes的公開輸出節點。app.build()驗證並編譯整個圖,然後傳回一個Run。run.push("image", ...)將資料傳送到指定的輸入節點。run.pull("classes", ...)從指定的輸出節點接收資料。
與 Python 程式碼相同的結構:
import pyneat
model = pyneat.Model("resnet50.tar.gz")
app = pyneat.Graph()
app.add(pyneat.nodes.input("image"))
app.add(model)
app.add(pyneat.nodes.output("classes"))
run = app.build()
image = ... # Create or load a tensor-compatible object.
run.push("image", [image])
result = run.pull("classes", timeout_ms=1000)
run.stop()
Python 的 Run.push(...) 函式需要一個類似批次(batch)的序列。請傳遞 [tensor] 或 [sample],而不是單獨的張量/樣本物件。
執行圖
一個建構好的 Run 接受與 Neat 中其他位置使用的相同類型的公用負載:
| 負載 | 何時使用 |
|---|---|
TensorList | 當您傳遞張量且不需要額外的樣本中繼資料時。 |
Sample | 當您需要時間戳、frame_id、stream_id、文字/音訊/視訊中繼資料、檢測結果或 EOS(序列結束)時。 |
std::vector<cv::Mat> | 當您想要從 OpenCV 取得方便的影像輸入時。 |
常見的 C++ 呼叫:
run.push(neat::TensorList{image});
run.push("image", neat::TensorList{image});
run.push(sample);
run.push("image", sample);
auto out = run.pull(/*timeout_ms=*/1000);
auto named = run.pull("classes", /*timeout_ms=*/1000);
neat::TensorList tensors = run.pull_tensors("classes", 1000);
neat::Sample sample_out = run.pull_samples("classes", 1000);
當逾時或關閉時,應回傳一個空的 std::optional,此時請使用 pull(...)。當您需要一個帶有類型且在逾時或發生錯誤時會引發異常的便利輔助函式時,請使用 pull_tensors(...) 或 pull_samples(...)。
對於有限的應用程式推送串流,在收集最終指標之前,請先關閉輸入並清空緩衝:
run.close_input();
while (auto out = run.pull("classes", 1000)) {
// Drain remaining output.
}
run.stop();
對於以任務為中心的執行階段設定指南,包括佇列策略、輸出所有權、資料丟棄遙測和多串流測量,請參閱 執行圖形。
build() 與 build(first_input)
大多數圖都可以無需輸入樣本的情況下建構:
neat::Run run = app.build();
當圖已經宣告足夠的形狀/邊界資訊時,或者當圖擁有其來源節點時(例如 RTSP/檔案/靜態影像輸入),請使用此設定。
「已設定初始值」建置會在建置期間為 Neat 提供第一個輸入:
neat::Run run = app.build(neat::TensorList{first_image});
當第一個輸入應該在開始串流之前,先設定形狀/格式的調整時,請使用此設定。預設情況下,已啟用「已設定的建置預先檢查」,因此 Neat 可以一次推送/提取設定,以在建置過程中捕捉第一個樣本的錯誤,而不是傳回一個 Run,該設定會在稍後立即導致失敗。
為了獲得更好的 輸送量、延遲和功耗資料,請在實際工作負載執行後儲存指標,而不是在建置後立即儲存。
圖的名稱不是端點名稱
Graph("name") 是一個用於診斷、儲存圖形檔案和視覺化的標籤。它不會宣告一個名為 name 的公開輸入或輸出。
錯誤的心理模型:
neat::Graph camera("image");
// This does not make run.push("image", ...) valid by itself.
正確的端點宣告:
neat::Graph camera("camera_route");
camera.add(neat::nodes::Input("image"));
至於輸出結果:
neat::Graph classifier("classifier");
classifier.add(neat::nodes::Output("classes"));
將 Input("image") 和 Output("classes") 視為圖的片段中公開的入口。圖的名稱就像建築物上的標誌。
檢查端點名稱,而不是猜測
在建構之前,請檢查圖中宣告的邏輯公開端點:
for (const auto& name : app.inputs()) {
std::cout << "graph input: " << name << "\n";
}
for (const auto& name : app.outputs()) {
std::cout << "graph output: " << name << "\n";
}
建置完成後,檢查 Run 實際上接受哪些內容:
for (const auto& name : run.input_names()) {
std::cout << "run input: " << name << "\n";
}
for (const auto& name : run.output_names()) {
std::cout << "run output: " << name << "\n";
}
將此用於模型路徑以及任何多輸入/多輸出應用程式。端點匹配必須精確:
Input("image_l") 可以繫結到名為 image_l 的模型輸入;Input("my_random_name") 則不行。
未命名的便利 API
對於單輸入/單輸出圖,您可以省略端點名稱:
neat::Graph app;
app.add(neat::nodes::Input());
app.add(model);
app.add(neat::nodes::Output());
neat::Run run = app.build();
run.push(neat::TensorList{image});
auto result = run.pull(1000);
這對於快速的腳本和測試非常方便。對於非簡單的應用程式,建議使用名稱。
如果一個圖有許多可能的輸入或輸出,未命名的 push(...) 或 pull() 會導致錯誤,並
回報可用的名稱。這種錯誤是故意的:Neat 不應該猜測您指的是哪個攝影機、張量或輸出層。
模型是圖的片段
一個 Model 可以直接新增到一個圖中:
neat::Model yolo("yolov8.tar.gz");
neat::Graph app;
app.add(neat::nodes::Input("image"));
app.add(yolo);
app.add(neat::nodes::Output("detections"));
Graph::add(model) 會將從檔案和模型選項中選取的模型路徑插入。該路徑可能包含預處理、MLA 推論、後處理、張量轉換和檢測解碼階段。
對於常見的線性情況,您不必手動呼叫 model.graph()。
對於更進階的組合,您可以檢查或重複使用該路徑作為 Graph 的一部分:
neat::Graph route = yolo.graph();
auto model_inputs = route.inputs();
auto model_outputs = route.outputs();
多輸入模型
對於多輸入模型,請勿猜測名稱。請詢問路線:
neat::Graph route = model.graph();
for (const auto& name : route.inputs()) {
std::cout << "model expects input: " << name << "\n";
}
然後,為您的上游片段命名,使其與模型的輸入名稱相符:
neat::Graph left_camera;
left_camera.add(neat::nodes::Input("image_l"));
neat::Graph uv_camera;
uv_camera.add(neat::nodes::Input("image_uv"));
neat::Graph app;
app.connect(left_camera, route); // Binds image_l -> model image_l.
app.connect(uv_camera, route); // Binds image_uv -> model image_uv.
如果宣告了 left_camera,並且指定了 Input("a_new_name_image_l"),它就不會與 image_l 綁定。請新增一個小型適配器圖,並使用正確的端點名稱,而不是依賴隱含的重新命名。
獨立模型圖
預設情況下,model.graph() 會傳回一個可重複使用的模型片段,其中包含開放的命名端點。如果您希望傳回的圖可以獨立執行,請要求明確的公開輸入/輸出節點:
neat::Model::RouteOptions route_opt;
route_opt.include_input = true;
route_opt.include_output = true;
neat::Graph standalone = model.graph(route_opt);
neat::Run run = standalone.build();
為了進階或除錯用途,模型路徑可以公開個別的實體輸出:
route_opt.expose_all_outputs = true;
除非您明確需要個別的實體輸出緩衝區,否則請將此選項停用。預設模型的行為是公開路由合約預期的邏輯模型輸出。如果模型只有一個實體輸出,expose_all_outputs = true 仍然只會公開一個輸出。
add() 與 connect()
有兩種組合工具:
| API | 意義 | 何時使用 |
|---|---|---|
add(x) | 將內容附加或插入到目前的線性鏈中。 | 您指的是「同一管線中的下一個步驟」。 |
connect(a, b) | 透過命名的端點連接兩個圖的片段。 | 您正在組合可重複使用的片段或建置拓撲結構。 |
connect("a", "b") | 連接已在同一圖中宣告的兩個端點。 | 您正在建置一個小型輔助片段。 |
線性組合:
neat::Graph app;
app.add(neat::nodes::Input("image"));
app.add(model);
app.add(neat::nodes::Output("classes"));
片段組成:
neat::Graph app;
app.connect(camera, model_route);
app.connect(model_route, output_sink);
輔助片段 內部的內部端點接線:
neat::Graph pass_through("pass_through");
pass_through.add(neat::nodes::Input("in"));
pass_through.add(neat::nodes::Output("out"));
pass_through.connect("in", "out");
主要規則:add() 表示一個線性鏈。connect() 表示圖的拓撲結構。
可重複使用的 Graph 片段
函式可以傳回可重複使用的 Graph 片段:
neat::Graph make_classifier(neat::Model& model) {
neat::Graph g("classifier");
g.add(neat::nodes::Input("image"));
g.add(model);
g.add(neat::nodes::Output("classes"));
return g;
}
以線性方式使用可重複使用的程式碼片段:
neat::Graph classifier = make_classifier(model);
neat::Graph app;
app.add(classifier);
或者明確地指定線段片段:
neat::Graph app;
app.connect(camera, classifier);
app.connect(classifier, class_sink);
如果將節點 add() 附加到一個分支之後會產生歧義,則 Neat 會失敗,並提示您改用 connect(...)。
這樣比默默地附加到錯誤的分支上更好。
分支單一資料流
當單一輸入資料流應該導向多個具名輸出時,請使用 graphs::Branch。
neat::Graph branch = neat::graphs::Branch("image", {"preview", "model_input"});
意義:
image -> preview
-> model_input
範例:
neat::Graph camera;
camera.add(neat::nodes::Input("image"));
neat::Graph preview;
preview.add(neat::nodes::Output("preview"));
neat::Graph branch = neat::graphs::Branch("image", {"preview", "model_input"});
neat::Graph app;
app.connect(camera, branch);
app.connect(branch, preview);
在將分支連接到模型時,請選擇與模型輸入名稱相符的分支輸出名稱:
neat::Graph route = model.graph();
for (const auto& name : route.inputs()) {
std::cout << "choose a branch output matching: " << name << "\n";
}
分支是明確的,因為它會影響佇列和反壓機制。如果某個分支速度較慢,根據輸出選項和下游圖,它可能會減慢速度或導致資料遺失,相對於其他分支而言。
Python:
branch = pyneat.graphs.branch("image", ["preview", "model_input"])
結合多個資料流
當需要將多個輸入資料流合併為單一邏輯輸出時,請使用 graphs::Combine。
neat::Graph pair = neat::graphs::Combine({"left", "right"},
"stereo",
neat::CombinePolicy::ByFrame);
意義:
left --\
+--> stereo
right --/
策略:
| 策略 | 意義 |
|---|---|
CombinePolicy::None | 不自動合併。多個來源指向單一輸出時,會導致輸出失敗。 |
CombinePolicy::ByFrame | 將樣本與完全相同的 Sample::frame_id 進行匹配。如果缺少畫面 ID,則會導致失敗;沒有 PTS 備用方案。 |
CombinePolicy::ByPts | 將樣本與完全相同的 Sample::pts_ns 呈現時間戳進行匹配。如果缺少 PTS,則會導致失敗;沒有畫面 ID 備用方案。 |
簡潔的說明:
ByFrame表示「給我具有相同畫面編號的左右樣本」。ByPts表示「給我具有相同媒體時間戳的樣本」。
範例:
neat::Graph left;
left.add(neat::nodes::Input("left"));
neat::Graph right;
right.add(neat::nodes::Input("right"));
neat::Graph pair = neat::graphs::Combine({"left", "right"},
"stereo",
neat::CombinePolicy::ByFrame);
neat::Graph app;
app.connect(left, pair);
app.connect(right, pair);
neat::Run run = app.build();
run.push("left", left_sample_with_frame_id_42);
run.push("right", right_sample_with_frame_id_42);
auto stereo = run.pull("stereo", 1000);
Python:
pair = pyneat.graphs.combine(["left", "right"], "stereo", pyneat.CombinePolicy.ByFrame)
如果樣本不包含所需的關鍵資訊,合併階段將會失敗,並顯示診斷訊息,而不是進行推測。