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 最能发挥价值的场景。
作者： Civil Learning