Daggr：介于 Gradio 和 ComfyUI 之間的 AI 工作流可視化方案

Daggr 是一個代碼優先的 Python 庫，可將 AI 工作流轉換為可視化圖，支持對 Gradio 管道進行檢查、重跑和調試。

單模型、單 prompt 的簡單 demo 通常不會有什么問題。但當工作流擴展到多個步驟，比如加入后處理函數、背景移除、轉錄摘要、檢索重排等等時情況就開始失控了。

狀態在各個環節之間流轉，我們不得不反復運行 cell、打印中間結果、注釋掉大段代碼來定位問題。每次出錯，甚至不確定該從哪個環節開始排查：是輸入有問題？模型出了狀況？還是中間的膠水代碼邏輯不對？

這種場景在 AI 應用開發中極為常見。

Daggr 正是為解決這類問題而設計的。它不是要取代 Python，也不是強推拖拽式編輯器，而是填補一個長期存在的空白：用代碼定義工作流，用可視化圖審視系統狀態。

Daggr 概述

Daggr 是一個用于構建 AI 工作流的開源 Python 庫。工作流通過代碼定義，使用標準 Python 語法，無需 DSL 或 YAML 配置。

Daggr 的核心功能是從代碼生成可視化畫布。這張畫布是一個實時更新、可交互檢查的有向圖，精確反映代碼的執行狀態。每個計算步驟對應一個節點，節點之間的數據流向清晰可見，所有中間輸出均可點擊查看、單獨重跑或回溯歷史。

一個關鍵的設計決策是：可視化層僅作為觀察工具，代碼始終是唯一的事實來源。這一選擇決定了 Daggr 與傳統可視化編排工具的本質區別。

使用體驗

安裝：

pip install daggr

創建一個 Python 文件，例如 app.py：

import random
import gradio as gr
from daggr import GradioNode, Graph
glm_image = GradioNode(
"hf-applications/Z-Image-Turbo",
api_name="/generate_image",
inputs={
"prompt": gr.Textbox(
label="Prompt",
value="A cheetah in the grassy savanna.",
lines=3,
),
"height": 1024,
"width": 1024,
"seed": random.random,
},
outputs={
"image": gr.Image(
label="Image"
),
},
)
background_remover = GradioNode(
"hf-applications/background-removal",
api_name="/image",
inputs={
"image": glm_image.image,
},
postprocess=lambda _, final: final,
outputs={
"image": gr.Image(label="Final Image"),
},
)
graph = Graph(
name="Transparent Background Image Generator", nodes=[glm_image, background_remover]
)
graph.launch()

運行：

daggr app.py

輸出的不是傳統的黑盒 Gradio demo，而是一張可視化畫布：兩個節點通過邊連接，輸入參數可調整，輸出結果可檢查。開發者可以單獨重跑圖像生成節點或背景移除節點，也可以在歷史結果之間切換，觀察下游節點如何響應不同的輸入狀態。

整個調試過程無需 print 語句，無需人工追蹤狀態變化。

與 Gradio 的差異

Gradio 在構建單步 demo 方面表現出色，但當工作流涉及多個步驟時，調試難度顯著上升。修改一個 prompt 后，下游某處出現問題，但難以確定：該步驟是否重新執行？使用的是哪組輸入參數？

Daggr 直接解決了這一問題。每次節點運行都會被記錄，每個輸出結果都可追溯其來源，每條連接都標記了數據的新鮮度。當上游值發生變化時，Daggr 會通過視覺提示告知開發者；下游節點若未重新執行，狀態一目了然。

Daggr 的工作流模型非常直觀：工作流本質上是一個有向無環圖（DAG）。

每個節點代表一次計算操作，可以是 Gradio Space API 調用、Hugging Face 推理請求，或普通的 Python 函數。節點通過輸入端口和輸出端口定義接口，數據沿著端口之間的連接流動。

核心概念就是這些，但實現細節中有許多值得關注的設計。

GradioNode：封裝現有 Gradio 應用

GradioNode 用于調用已有的 Gradio 應用，支持 Hugging Face Spaces 上的遠程應用和本地運行的應用。

from daggr import GradioNode
import gradio as gr
image_gen = GradioNode(
space_or_url="black-forest-labs/FLUX.1-schnell",
api_name="/infer",
inputs={
"prompt": gr.Textbox(label="Prompt"),
"seed": 42,
"width": 1024,
"height": 1024,
},
outputs={
"image": gr.Image(label="Generated Image"),
},
)

對于熟悉 Hugging Face Spaces "Use via API" 功能的開發者，這種接口定義方式會非常熟悉。Daggr 采用了相同的參數命名和端點定義規范。

由于 GradioNode 調用的是外部服務，默認采用并發執行模式，無需處理線程管理或鎖機制。

FnNode：自定義 Python 函數

當工作流需要自定義邏輯而非模型調用時，FnNode 提供了相應的支持。典型應用場景包括數據解析、過濾、組合和后處理。

from daggr import FnNode
import gradio as gr
def summarize(text: str, max_words: int = 100) -> str:
words = text.split()[:max_words]
return " ".join(words) + "..."
summarizer = FnNode(
fn=summarize,
inputs={
"text": gr.Textbox(label="Text to Summarize", lines=5),
"max_words": gr.Slider(minimum=10, maximum=500, value=100),
},
outputs={
"summary": gr.Textbox(label="Summary"),
},
)

Daggr 會自動檢查函數簽名，按名稱匹配輸入參數，按順序將返回值映射到輸出端口。

值得注意的是，FnNode 默認采用串行執行模式。這是一個經過權衡的設計決策：本地 Python 代碼可能涉及文件操作、GPU 資源、全局狀態，以及各種非線程安全的庫。Daggr 選擇了更保守的默認行為。

如需并發執行，可以顯式聲明：

node = FnNode(my_func, concurrent=True)

InferenceNode：云端模型推理

InferenceNode 允許通過推理服務直接調用 Hugging Face 模型，無需下載模型權重或配置本地環境。

from daggr import InferenceNode
import gradio as gr
llm = InferenceNode(
model="meta-llama/Llama-3.1-8B-Instruct",
inputs={
"prompt": gr.Textbox(label="Prompt", lines=3),
},
outputs={
"response": gr.Textbox(label="Response"),
},
)

InferenceNode 默認并發執行，并自動傳遞 Hugging Face token，支持 ZeroGPU 計費追蹤、私有 Space 訪問和受限模型調用。

Daggr 一些主要特征

溯源是 Daggr 的核心特性之一。

每次節點執行時，Daggr 都會保存輸出結果及產生該結果的精確輸入參數。結果歷史可以像版本控制一樣瀏覽。選擇某個歷史結果時，Daggr 會自動恢復當時的輸入狀態，不僅針對當前節點，下游節點的狀態也會同步恢復。

這意味著開發者可以自由探索不同的參數變體而不丟失上下文。例如，生成三張圖片，對其中兩張執行背景移除，之后選擇第一張圖片，整個工作流圖會自動對齊到對應的狀態。

這不僅僅是便利性的提升，而是一種不同的開發范式。

狀態可視化

Daggr 使用邊的顏色傳遞數據狀態信息：橙色表示數據是最新的，灰色表示數據已過期。

當上游輸入發生變化時，所有依賴該輸入的邊都會變為灰色，清晰地指示哪些節點需要重新執行。

Scatter 和 Gather 模式

部分工作流需要處理列表數據：生成多個項目，分別處理，最后合并結果。Daggr 通過 .each 和 .all() 語法支持這種模式：

script = FnNode(fn=generate_script, inputs={...}, outputs={"lines": gr.JSON()})
tts = FnNode(
fn=text_to_speech,
inputs={
"text": script.lines.each["text"],
"speaker": script.lines.each["speaker"],
},
outputs={"audio": gr.Audio()},
)
final = FnNode(
fn=combine_audio,
inputs={"audio_files": tts.audio.all()},
outputs={"audio": gr.Audio()},
)

語法仍然是標準 Python，邏輯顯式清晰，同時 Daggr 能夠理解數據的分發與聚合語義。

Choice 節點

當需要在多個備選方案之間切換時，例如使用不同的圖像生成器或 TTS 服務，但保持下游邏輯不變，可以使用 Choice 節點：

host_voice = GradioNode(...) | GradioNode(...)

UI 中會顯示一個選擇器，下游連接保持不變，選擇結果在 sheet 中持久保存。這種設計便于進行對比實驗，同時保持代碼庫的整潔。

Sheets：多狀態工作區

Daggr 引入了 sheets 的概念，可以理解為獨立的工作區。每個 sheet 擁有獨立的輸入參數、緩存結果和畫布布局，但共享相同的工作流定義。

這與復制 notebook 進行實驗的場景類似，但管理更加規范。

API 與部署

Daggr 工作流自動暴露 REST API，可以通過以下方式查詢 schema：

curl http://localhost:7860/api/schema

部署同樣簡潔：

daggr deploy my_app.py

Daggr 會自動提取工作流圖、創建 Hugging Face Space、生成元數據并完成部署。

總結

對于單模型 demo，Gradio 已經足夠；對于純可視化編排需求，ComfyUI 可能更合適；對于生產級任務調度，Airflow 或 Prefect 是更成熟的選擇。

而Daggr 的定位是中間地帶：工作流復雜度足以需要可視化檢查和調試，但尚未達到需要正式編排系統的程度；開發者仍處于探索、調整和迭代的階段。

這是 Daggr 最能發揮價值的場景。

https://avoid.overfit.cn/post/725b46b7dd434d9eb3a90ff9d67b968a

作者： Civil Learning