Daggr

Gradio团队正式开源了一款专为AI工作流构建设计的Python库——Daggr。这款工具以“代码优先”为核心，能够无缝连接Gradio应用、Hugging Face推理模型与自定义Python函数，还能自动生成可视化画布，支持分步调试、状态持久化，彻底解决复杂AI流程串联难、调试繁琐、中间结果丢失的痛点，仅需几行Python代码即可搭建高效、可追溯的AI工作流。

为什么需要Daggr？解决复杂AI工作流的核心痛点

对于经常构建多模型、多步骤串联的AI应用开发者而言，想必都遭遇过这些困境：

1. 流程串联繁琐：需要手动串联多个API调用，代码耦合度高，一旦某个步骤变更，整个脚本都可能需要重构；
2. 调试效率低下：10步工作流若第5步出现问题，往往需要重新运行全部步骤，才能定位问题所在，耗时耗力；
3. 中间结果丢失：缺乏有效的状态保存机制，中断后无法恢复此前的运行结果，只能重新开始；
4. 工具选型尴尬：要么编写脆弱的脚本，难以维护和版本控制；要么使用重型生产级编排平台，学习成本高，不适合快速实验。

Daggr的出现，正是为了填补“快速实验”与“稳定编排”之间的空白，它聚焦于AI工作流的快速搭建与高效调试，兼顾代码的可版本控制与可视化的便捷性。

Daggr核心特性：代码优先，兼顾便捷与可控

1. 代码优先+自动可视化，两全其美

Daggr摒弃了传统节点式拖拽编辑的方式，采用代码优先的设计理念：

• 开发者用纯Python代码定义工作流节点与连接关系，支持Git版本控制，方便团队协作与代码回溯；
• 代码运行后，自动生成交互式可视化画布，清晰展示整个工作流的节点关系、输入输出，无需手动绘制；
• 可视化画布不仅是“展示面板”，更是“调试面板”，可直接查看任意节点的中间输出，直观定位问题。

2. 分步调试与重跑，大幅提升效率

这是Daggr最核心的实用价值，彻底解决“全流程重跑”的痛点：

• 支持单独查看任意节点的输入与输出，无需运行整个工作流，快速定位异常节点；
• 修改某个节点的输入或配置后，可单独重跑该节点（及后续依赖节点），无需从头开始，尤其适合长流程、耗时久的AI任务（如图像生成、3D建模）；
• 支持配置“备用节点”，可快速用一个模型或Gradio Space替换另一个，搭建具有容错性的弹性工作流，方便对比不同模型的效果。

3. 无缝集成Gradio，开箱即用

作为Gradio团队的作品，Daggr与Gradio生态实现深度兼容，无需额外适配：

• 直接引用公共或私有Gradio Space的名称与API端点，即可将其作为工作流中的节点，无需编写适配器或包装器；
• 支持run_locally=True参数，自动克隆Gradio Space，创建隔离虚拟环境并本地启动运行，执行失败时会优雅回退到远程API；
• 生成的工作流可直接通过Gradio的隧道服务生成公共分享链接，方便团队协作测试。

4. 状态持久化，支持多工作空间

Daggr会自动保存整个工作流的运行状态，避免中断后前功尽弃：

• 持久化内容包括：输入值、节点缓存结果、可视化画布位置、运行日志等，重启后可从上次停止的地方继续；
• 支持“工作表”功能，可在同一个应用中维护多个独立工作空间，分别处理不同的任务场景，互不干扰。

快速上手：安装与极简示例

1. 环境要求与安装

Daggr要求Python 3.10及以上版本，支持pip或uv安装，步骤简单：

# pip 安装
pip install daggr

# 或 uv 安装（更快更稳定）
uv pip install daggr

2. 极简示例：生成图像并移除背景

这个示例将串联两个Gradio Space，实现“文本生成图像→移除图像背景”的工作流，运行后自动生成可视化画布：

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

# 节点1：调用Gradio Space生成图像
image_gen = GradioNode(
    "hf-applications/Z-Image-Turbo",  # Gradio Space名称
    api_name="/generate_image",  # Space的API端点
    inputs={
        "prompt": gr.Textbox(
            label="生成图像提示词",
            value="一只猎豹在草原上疾驰。",
            lines=3,
        ),
        "height": 1024,
        "width": 1024,
        "seed": random.random,  # 随机种子
    },
    outputs={
        "image": gr.Image(label="生成的原始图像"),
    },
)

# 节点2：调用Gradio Space移除图像背景（连接节点1的输出作为输入）
bg_remover = GradioNode(
    "hf-applications/background-removal",
    api_name="/image",
    inputs={
        "image": image_gen.image,  # 直接关联前一个节点的输出，实现流程串联
    },
    outputs={
        "original_image": None,  # 隐藏该输出，不显示在可视化画布中
        "final_image": gr.Image(label="移除背景后的最终图像"),
    },
)

# 构建工作流并启动
graph = Graph(
    name="透明背景图像生成器", 
    nodes=[image_gen, bg_remover]  # 传入所有节点，自动识别连接关系
)
graph.launch()  # 默认在端口7860启动，打开浏览器即可访问可视化画布

运行上述脚本后，打开浏览器访问http://localhost:7860，即可看到可视化工作流画布，可修改提示词、查看每个节点的输出，还能单独重跑某个节点，无需重新生成整个流程。

3. 生成公共分享链接

若需要分享给团队成员测试，可添加share=True参数，生成临时公共URL：

graph.launch(share=True)

三种核心节点类型：覆盖各类AI工作流场景

Daggr支持三种节点类型，可灵活组合，满足不同的任务需求：

1. GradioNode：调用Gradio Space/本地Gradio应用

用于调用远程Gradio Space或本地部署的Gradio应用，是Daggr的核心节点之一，示例如下：

# 远程调用Gradio Space
remote_node = GradioNode(
    "username/your-space-name",
    api_name="/predict",
    inputs={"text": gr.Textbox(label="输入文本")},
    outputs={"result": gr.Textbox(label="输出结果")},
)

# 本地克隆并运行Gradio Space（失败自动回退远程）
local_node = GradioNode(
    "hf-applications/background-removal",
    api_name="/image",
    run_locally=True,  # 开启本地运行
    inputs={"image": gr.Image(label="输入图像")},
    outputs={"final_image": gr.Image(label="输出图像")},
)

2. FnNode：运行自定义Python函数

用于嵌入自定义的Python逻辑，如数据处理、格式转换等，示例如下：

from daggr import FnNode

# 自定义处理函数：将文本转为大写
def text_upper_process(text: str) -> str:
    return text.upper()

# 封装为FnNode节点
fn_node = FnNode(
    fn=text_upper_process,  # 传入自定义函数
    inputs={"text": gr.Textbox(label="原始文本")},
    outputs={"result": gr.Textbox(label="大写文本")},
)

3. InferenceNode：调用Hugging Face推理模型

用于通过Hugging Face推理服务提供商调用各类预训练模型，无需本地部署，示例如下：

from daggr import InferenceNode

# 调用HF推理服务中的模型
inference_node = InferenceNode(
    model="moonshotai/Kimi-K2.5:novita",  # 模型名称
    inputs={"prompt": gr.Textbox(label="推理提示词")},
    outputs={"response": gr.Textbox(label="模型响应结果")},
)

端到端实战：从图像到3D资产生成

下面以“图像→移除背景→缩小图像→增强3D风格→生成3D资产”为例，展示不同节点的组合使用，完整流程可直接运行：

from typing import Any
from PIL import Image
import uuid
import gradio as gr
from daggr import FnNode, GradioNode, InferenceNode, Graph
from daggr.state import get_daggr_files_dir

# 步骤1：本地运行Gradio Space，移除图像背景
background_remover = GradioNode(
    "merve/background-removal",
    api_name="/image",
    run_locally=True,
    inputs={"image": gr.Image(label="输入原始图像")},
    outputs={
        "original_image": None,
        "final_image": gr.Image(label="移除背景后的图像"),
    },
)

# 步骤2：自定义函数缩小图像，封装为FnNode
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)

downscaler = FnNode(
    downscale_image_to_file,
    name="图像缩小（提升后续推理效率）",
    inputs={
        "image": background_remover.final_image,
        "scale": gr.Slider(
            label="缩小比例",
            minimum=0.25,
            maximum=0.75,
            step=0.05,
            value=0.25,
        ),
    },
    outputs={"image": gr.Image(label="缩小后的图像", type="filepath")},
)

# 步骤3：调用HF推理模型，增强3D风格，封装为InferenceNode
flux_enhancer = InferenceNode(
    model="black-forest-labs/FLUX.2-klein-4B:fal-ai",
    inputs={
        "image": downscaler.image,
        "prompt": gr.Textbox(
            label="3D风格提示词",
            value="将此图像转换为清晰的3D资产渲染图，细节丰富，光影自然",
            lines=3,
        ),
    },
    outputs={"image": gr.Image(label="3D风格增强图像")},
)

# 步骤4：调用Gradio 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资产（GLB格式预览）")},
)

# 构建并启动完整工作流
graph = Graph(
    name="图像转3D资产工作流",
    nodes=[background_remover, downscaler, flux_enhancer, trellis_3d],
)

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

运行说明

1. 本地运行需登录Hugging Face Hub（执行huggingface-cli login）；
2. 部分Gradio Space在CPU上运行耗时较长（约10-20秒），有NVIDIA GPU可替换为支持CUDA的Space；
3. 运行后可在可视化画布中分步查看每个节点的输出，单独调整某个步骤的参数并重跑。

部署与后续展望

1. 永久部署

若需要长期托管工作流，可将其部署到Hugging Face Spaces，步骤如下：

1. 编写app.py（即上述工作流代码）；
2. 创建requirements.txt，添加daggr及其他依赖（如gradio、Pillow）；
3. 通过Hugging Face Hub上传文件，选择“Gradio”作为空间类型，即可完成部署。

2. 后续展望

目前Daggr处于Beta测试阶段，保持轻量易用的核心定位，未来可能会有以下更新：

• 优化API稳定性，减少版本迭代中的数据丢失风险；
• 扩展更多节点类型与第三方工具集成；
• 增强工作流的批量处理与调度能力。

若在使用过程中发现Bug或有功能需求，可通过Daggr官方GitHub仓库提交Issue，同时也可分享自己的工作流作品，有机会获得Gradio官方推荐。