Gradio 团队开源 Daggr：用 Python 链接应用并可视化检视

TL;DR： Daggr 是一个全新的开源 Python 库，用于构建连接 Gradio 应用、ML 模型和自定义函数的 AI 工作流。它会自动生成可视化画布，让你可以检查中间输出、重新运行单个步骤，并为复杂管道管理状态，这一切只需几行 Python 代码！

背景

如果你构建过结合多个模型或处理步骤的 AI 应用，你就知道那种痛苦：串联 API 调用、调试管道、丢失中间结果的踪迹。当一个 10 步工作流的第 5 步出错时，你往往不得不重新运行整个流程，才能看清发生了什么。

大多数开发者要么编写难以调试的脆弱脚本，要么转向为生产管道设计的重型编排平台——而不是用于快速实验。

我们一直在开发 Daggr，以解决我们在构建 AI 演示和工作流时反复遇到的问题：

可视化你的代码流程：与基于节点的 GUI 编辑器（通过拖拽和连接节点进行可视化操作）不同，Daggr 采用代码优先的方式。你用 Python 定义工作流，可视化画布会自动生成。这意味着你可以兼得两者之长：可版本控制的代码和对中间输出的可视化检查。

检查并重新运行任意步骤：可视化画布不只是摆设。你可以检查任意节点的输出、修改输入，并重新运行单个步骤，而无需执行整个管道。当你调试一个 10 步工作流、而只有第 7 步出问题时，这一点非常宝贵。你甚至可以提供“备用节点”——用一个模型或 Space 替换另一个——来构建具有韧性的工作流。

一流的 Gradio 集成：由于 Daggr 由 Gradio 团队构建，它与 Gradio Spaces 无缝协作。指向任何公共（或私有）Space，你就可以将其用作工作流中的节点。无需适配器，无需包装器——只需引用 Space 名称和 API 端点。

状态持久化：Daggr 会自动保存你的工作流状态、输入值、缓存结果、画布位置——这样你就可以从上次中断的地方继续。使用“sheets”在同一个应用内维护多个工作区。

快速开始

使用 pip 或 uv 安装 daggr，它只需要 Python 3.10 或更高版本：

pip install daggr
uv pip install daggr

下面是一个简单的示例，它生成一张图像并移除其背景。请查看 Space 底部的这个 Space 的 API 参考，了解它接受哪些输入、产生哪些输出。在这个示例中，该 Space 同时返回原始图像和编辑后的图像，因此我们只返回编辑后的图像。

import random
import gradio as gr
from daggr import GradioNode, Graph

# Generate an image using a Gradio Space
image_gen = GradioNode(
    "hf-applications/Z-Image-Turbo",
    api_name="/generate_image",
    inputs={
        "prompt": gr.Textbox(
            label="Prompt",
            value="A cheetah sprints across the grassy savanna.",
            lines=3,
        ),
        "height": 1024,
        "width": 1024,
        "seed": random.random,
    },
    outputs={
        "image": gr.Image(label="Generated Image"),
    },
)

# Remove background using another Gradio Space
bg_remover = GradioNode(
    "hf-applications/background-removal",
    api_name="/image",
    inputs={
        "image": image_gen.image,  # Connect to previous node's output
    },
    outputs={
        "original_image": None,  # Hide this output
        "final_image": gr.Image(label="Final Image"),
    },
)

graph = Graph(
    name="Transparent Background Generator", 
    nodes=[image_gen, bg_remover]
)
graph.launch()

就是这样。运行这个脚本，你会得到一个自动启动、在 7860 端口上提供的可视化画布，以及一个可分享的实时链接，其中显示两个节点已连接，你可以在每一步修改输入并检查输出。

节点类型

Daggr 支持三种类型的节点：

GradioNode 调用 Gradio Space API 端点或本地提供的 Gradio 应用。传入 run_locally=True，Daggr 会自动克隆该 Space、创建一个隔离的虚拟环境并启动该应用。如果本地执行失败，它会优雅地回退到远程 API。

node = GradioNode(
    "username/space-name",
    api_name="/predict",
    inputs={"text": gr.Textbox(label="Input")},
    outputs={"result": gr.Textbox(label="Output")},
)

# clone a Space locally and serve
node = GradioNode(
    "hf-applications/background-removal",
    api_name="/image",
    run_locally=True,
    inputs={"image": gr.Image(label="Input")},
    outputs={"final_image": gr.Image(label="Output")},

FnNode — 运行自定义 Python 函数：

def process(text: str) -> str:
    return text.upper()

node = FnNode(
    fn=process,
    inputs={"text": gr.Textbox(label="Input")},
    outputs={"result": gr.Textbox(label="Output")},
)

InferenceNode — 通过 HF Mirror Inference Providers 调用模型：

node = InferenceNode(
    model="moonshotai/Kimi-K2.5:novita",
    inputs={"prompt": gr.Textbox(label="Prompt")},
    outputs={"response": gr.Textbox(label="Response")},
)

共享你的工作流

使用 Gradio 的隧道生成公共 URL：

graph.launch(share=True)

如需永久托管，请使用 Gradio SDK 部署到 HF Mirror Spaces——只需将 daggr 添加到你的 requirements.txt。

使用不同节点的端到端示例

我们现在将开发一个应用，接收一张图像并生成一个 3D 资产。此演示可在 daggr 0.4.3 上运行。步骤如下：

接收一张图像，移除背景：为此，我们将克隆 BiRefNet Space 并在本地运行它。
为提高效率而缩小图像尺寸：我们将使用 FnNode 为此编写一个简单的函数。
生成 3D 资产风格的图像以获得更好的效果：我们将在 Inference Providers 上使用 InferenceNode 搭配 Flux.2-klein-4B 模型。
将输出图像传递给 3D 生成器：我们将把输出图像发送到托管在 Spaces 上的 Trellis.2 Space。

本地运行的 Spaces 可能会在应用程序文件中将模型加载到 CUDA（使用 to.(“cuda”)）或 ZeroGPU。要禁用此行为以在 CPU 上运行模型（如果你有没有 NVIDIA GPU 的设备，这会很有用），请复制你想使用的 Space 并克隆它。

生成的图如下所示。

让我们编写第一步，即背景移除器。我们将在本地克隆并运行这个 Space。这个 Space 在 CPU 上运行，大约需要 13 秒。如果你有 NVIDIA GPU，可以换成这个应用。

from daggr import FnNode, GradioNode, InferenceNode, Graph

background_remover = GradioNode(
   "merve/background-removal",
   api_name="/image",
   run_locally=True, 
   inputs={
       "image": gr.Image(),
   },
   outputs={
       "original_image": None,
       "final_image": gr.Image(
           label="Final Image"
       ),
   },
)

对于第二步，我们需要编写一个辅助函数来缩小图像，并将其传递给 FnNode。

from PIL import Image
from daggr.state import get_daggr_files_dir


def downscale_image_to_file(image: Any, scale: float = 0.25) -> str | None:
   pil_img = Image.open(image)
   scale_f = max(0.05, min(1.0, float(scale)))
   w, h = pil_img.size
   new_w = max(1, int(w * scale_f))
   new_h = max(1, int(h * scale_f))
   resized = pil_img.resize((new_w, new_h), resample=Image.LANCZOS)
   out_path = get_daggr_files_dir() / f"{uuid.uuid4()}.png"

   resized.save(out_path)
   return str(out_path)

我们现在可以将该函数传入以初始化 FnNode。

downscaler = FnNode(
   downscale_image_to_file,
   name="Downscale image for Inference",
   inputs={
       "image": background_remover.final_image,
       "scale": gr.Slider(
           label="Downscale factor",
           minimum=0.25,
           maximum=0.75,
           step=0.05,
           value=0.25,
       ),
   },
   outputs={
       "image": gr.Image(label="Downscaled Image", type="filepath"),
   },
)

我们现在将使用 Flux 模型编写 InferenceNode。

flux_enhancer = InferenceNode(
   model="black-forest-labs/FLUX.2-klein-4B:fal-ai",
   inputs={
       "image": downscaler.image,
       "prompt": gr.Textbox(
           label="prompt",
           value=("Transform this into a clean 3D asset render"),
           lines=3,
       ),
   },
   outputs={
       "image": gr.Image(label="3D-Ready Enhanced Image"),
   },
)

将使用 InferenceNode 的应用部署到 HF Mirror Spaces 时，请使用细粒度的 HF Mirror 访问令牌，并且只勾选“Make calls to Inference Providers”选项。

最后一个节点是通过查询 HF Mirror 上的 Trellis.2 Space 进行 3D 生成。

trellis_3d = GradioNode(
   "microsoft/TRELLIS.2",
   api_name="/image_to_3d",
   inputs={
       "image": flux_enhancer.image,
       "ss_guidance_strength": 7.5,   
       "ss_sampling_steps": 12,     
   },
   outputs={
       "glb": gr.HTML(label="3D Asset (GLB preview)"),
   },
)

将它们串联起来并启动应用就像下面这样简单。

graph = Graph(
   name="Image to 3D Asset Pipeline",
   nodes=[background_remover, downscaler, flux_enhancer, trellis_3d],
)

if __name__ == "__main__":
   graph.launch()

你可以在这个 Space 中找到完整示例的运行，若要在本地运行，你只需获取 app.py、安装依赖项并登录 HF Mirror Hub。

下一步

Daggr 处于测试阶段，并且有意保持轻量。API 可能会在版本之间发生变化，虽然我们会在本地持久化工作流状态，但更新期间可能会发生数据丢失。如果你有功能请求或发现 bug，请在此处提交 issue。我们期待你的反馈！在社交媒体上使用 Gradio 分享你的 daggr 工作流，有机会被推荐。查看所有精选作品请点击此处。

Gradio 团队开源 Daggr：用 Python 链接应用并可视化检视

目录

背景

快速开始

节点类型

共享你的工作流

使用不同节点的端到端示例

下一步