Daggr：用代码生成工作流画布

Daggr是一个代码优先的Python库，它可以将AI工作流转换为可视化图形。无需猜测，即可检查、重新运行和调试Gradio流水线。

你从一个模型开始。一个提示词。一个简洁的小demo。生活很美好。

然后你添加了第二步。也许是一个后处理函数。也许是一个背景移除器。或者是转录后的摘要生成器。或者是检索后的重新排序器。

突然间，你需要在脑海中管理各种状态。

你在重新运行单元格。打印中间输出。注释掉一半的代码只是为了看看发生了什么变化。每次出现问题时，你甚至不确定应该先从哪里找起。

是输入的问题？ 模型的问题？ 还是中间的粘合代码的问题？

你知道那种感觉。就是那种"我发誓五分钟前它还工作正常"的感觉。

Daggr就是为了那一刻而存在的。

不是为了替代Python。不是为了强迫你使用拖拽式编辑器。而是为了给你提供我们长期以来一直在寻找的东西：一种像程序员一样编写工作流，并像系统一样查看它们的方式。

让我解释一下我的意思。

1、Daggr到底是什么

Daggr是一个用于构建AI工作流的开源Python库。

你在代码中定义工作流。普通的Python代码。没有DSL。没有YAML金字塔。

然后Daggr做了一件 quietly 强大的事情：它从代码生成一个可视化画布。

不是一个设计工具。 不是你需要时刻照看的节点编辑器。

而是一个实时的、可检查的图形，它精确地反映了你的代码在做什么。

每一步都变成一个节点。 每个连接都变得可见。 每个中间输出都变成了你可以点击、重新运行和回放的东西。

这是重要的一部分。

可视化层不是真理的来源。 你的代码才是。

这个设计选择改变了一切。

2、试一试：从代码到画布

安装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，而是一个画布。

两个节点。 它们之间有一条可见的边。 你可以调整的输入。 你可以检查的输出。

你可以只重新运行图像生成器。或者只重新运行背景移除器。或者切换到之前的结果，并观察整个下游图形自动更新以匹配该确切的历史记录。

没有打印语句。没有心理记账。

这就是Daggr的核心思想。

3、为什么这感觉不同

Gradio对于demo来说非常棒。但是一旦工作流变成多步骤，调试就变得……模糊不清。

你修改了一个提示词。 下游的东西坏了。 你不确定它是否重新运行了。 你不确定使用的是哪些输入。

Daggr直面这个问题。

每次节点运行都会被跟踪。 每个结果都有来源记录。 每个连接都知道它是新鲜的还是过时的。

如果上游的值发生了变化，Daggr会可视化地提醒你。 如果下游节点没有重新运行，你可以立即看到。

4、Daggr如何看待工作流

在其核心，Daggr非常简单。

工作流是一个有向图。

每个节点代表一个计算：

• Gradio Space API调用
• Hugging Face推理调用
• 或者普通的Python函数

节点有输入端口和输出端口。

端口是数据流动的方式。

就这样。

但是，魔鬼总是在细节中。

5、GradioNode：无痛封装现有应用

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"),
    },
)

如果你曾经在Space上点击过"Use via API"，这会感觉很熟悉。Daggr使用相同的参数名称和端点定义。

没有什么神奇之处。只是结构化。

而且因为它是外部的，GradioNodes默认并发运行。没有线程麻烦。没有锁。没有意外。

6、FnNode：你自己的代码所在的地方

有时候你不需要模型。你需要逻辑。

解析。过滤。组合。后处理。

这就是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检查函数签名。按名称匹配输入。按顺序将返回值映射到输出。

这里有一个微妙但重要的选择。

FnNodes默认是顺序的。

为什么。因为本地Python代码很乱。它操作文件。GPU。全局状态。不是线程安全的库。

Daggr倾向于安全的一面。

如果你知道一个函数可以并行运行，你可以选择加入：

node = FnNode(my_func, concurrent=True)

你总是掌控着一切。

7、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"),
    },
)

这些节点默认是并发的，并且会自动为以下内容传递你的Hugging Face令牌：

• ZeroGPU跟踪
• 私有Spaces
• 门控模型

再次说明，减少管道工作。更多思考。

8、真正有效的来源记录

这就是Daggr悄悄领先的地方。

每次节点运行时，Daggr都会存储：

• 输出
• 以及产生它的确切输入

你可以像浏览版本一样浏览结果历史。

当你选择一个旧的结果时，Daggr会自动恢复输入。

不只是针对那个节点。 也针对下游节点。

这意味着你可以在不丢失上下文的情况下探索变体。

生成三张图片。 移除其中两张的背景。 稍后再回来。 点击图片#1。

整个图形会重新回到对齐状态。

这不是一个便利功能。这是一个不同的心智模型。

9、可视化的陈旧状态

Daggr使用边颜色来展示真相。

橙色边表示新鲜。 灰色边表示陈旧。

如果你在上游更改了输入，每个依赖的边都会变灰。

没有猜测。没有"这个重新运行了吗？"

你看得见。

这听起来很小，直到你尝试它。然后你很难再回头。

10、分散和聚集：当列表变成流水线时

有些工作流自然会产生分支。

你生成多个项目。 然后处理每一个。 然后组合它们。

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。仍然是显式的。但是图形理解你的意思。

11、选择节点

有时候你需要替代方案。

两个图像生成器。 两个TTS提供商。 相同的下游逻辑。

Daggr允许你这样做：

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

在UI中，你会得到一个选择器。下游连接保持完整。

你的选择会持久化到每个sheet。

这对于实验来说非常完美，而不会让你的代码库分支变得混乱。

12、Sheets：平行宇宙，一个工作流

Daggr引入了sheets的概念。

把它们想象成工作空间。

每个sheet都有：

• 它自己的输入
• 它自己的缓存结果
• 它自己的画布位置

相同的工作流。不同的状态。

如果你曾经为了尝试新东西而复制过notebook，这会感觉很熟悉。而且更干净。

13、本地、远程，或者两者兼有

Daggr可以与：

• 远程Gradio Spaces
• Hugging Face推理
• 完全本地的应用

一起工作

你甚至可以通过设置run_locally=True在本地运行Space。

Daggr自动处理克隆、环境和回退。

如果失败了，它会准确地告诉你哪里出了问题以及原因。

14、API、部署和分享

Daggr工作流会自动暴露REST API。

你可以检查模式：

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

并以编程方式调用工作流。

部署只需要一个命令：

daggr deploy my_app.py

Daggr提取图形，设置Space，生成元数据，并为你推送所有内容。

没有繁琐的流程。

15、什么时候Daggr不是合适的工具

Daggr并不想做所有事情。

如果你想要：

• 单个模型demo - 使用Gradio
• 完全可视化的编辑器 - 使用ComfyUI
• 生产级调度器 - 使用Airflow或Prefect

Daggr存在于混乱的中间地带。

在那里，工作流足够复杂以至于需要检查。 但又足够流畅，你仍在思考、调整、探索。

这就是它的最佳用例。

原文链接: Daggr: Code-First Python Library That Turns AI Workflows Into Visual Graphs

汇智网翻译整理，转载请标明出处