用 Gradio Workflow 重建 AUTOMATIC1111：先拆工作流，再复刻界面

预计阅读时间：9 分钟

重建 AUTOMATIC1111 这类 Stable Diffusion Web UI，难点并不只是摆出提示词输入框、采样器下拉框和生成按钮。真正需要复刻的是一条可组合、可观察、能够持续扩展的推理工作流：参数进入系统，经过模型加载、采样、后处理，再把图像和元数据送回界面。

从来源标题能确认的核心方向，是使用 Gradio Workflow 重新构建 AUTOMATIC1111。由于摘要没有给出具体 API 或实现细节，下面把它当作一种架构实践来分析，并用标准 Gradio Blocks 给出一个可运行的最小项目。后续可以把其中的占位生成器替换成 Diffusers、ComfyUI API 或现有 Stable Diffusion 服务。

不要把界面事件直接写成推理系统

一个常见实现是把全部逻辑塞进按钮的点击回调：加载模型、解析提示词、生成图片、保存文件和更新历史记录都由同一个函数处理。原型阶段很快，但加入 LoRA、ControlNet、高清修复或批量任务后，回调会迅速失控。

更合适的划分是四层：

界面层：Gradio 组件、事件绑定和状态展示。
工作流层：定义步骤顺序、输入输出以及失败边界。
推理层：封装模型、采样器、设备和精度策略。
基础设施层：队列、缓存、文件存储、日志和任务取消。

例如，一个简化的 txt2img 流程可以表达为：

validate_inputs
    -> normalize_prompt
    -> resolve_model
    -> run_sampler
    -> postprocess
    -> persist_metadata
    -> render_result

这种拆分的价值在于，每个节点都能单独测试。更换模型后端时，Gradio 页面不需要跟着重写；增加一个后处理步骤时，也不用侵入采样器实现。

用明确的数据契约连接节点

工作流不应靠不断增长的参数列表传递上下文。可以定义一个任务对象，让每个节点读取和更新约定字段：

from dataclasses import dataclass, field
from typing import Any

@dataclass
class GenerationJob:
    prompt: str
    negative_prompt: str = ""
    width: int = 512
    height: int = 512
    steps: int = 20
    cfg_scale: float = 7.0
    seed: int = -1
    images: list[Any] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

边界处应尽早校验宽高、步数、模型名称和文件类型。GPU 推理开始后才发现宽度非法，不仅浪费时间，也可能占住显存和队列槽位。

还要把“用户输入参数”和“运行时状态”分开。提示词、尺寸和 seed 属于可复现参数；当前进度、GPU 设备和缓存命中情况属于运行状态。二者混在一起会让历史任务难以重放。

一个可运行的 Gradio 工作流骨架

下面的示例不下载模型，而是用 Pillow 生成确定性的占位图，因此可以直接运行。它展示了参数校验、任务构建、执行节点、元数据输出和 Gradio 队列。要接入真实模型，只需替换 generate_placeholder，保持输入输出契约不变。

创建 requirements.txt：

gradio>=4.44,<6
Pillow>=10,<12

创建 app.py：

import hashlib
import json
import random
from dataclasses import asdict, dataclass

import gradio as gr
from PIL import Image, ImageDraw


@dataclass
class GenerationJob:
    prompt: str
    negative_prompt: str
    width: int
    height: int
    steps: int
    cfg_scale: float
    seed: int


def build_job(prompt, negative_prompt, width, height, steps, cfg_scale, seed):
    prompt = prompt.strip()
    if not prompt:
        raise gr.Error("Prompt cannot be empty")

    width, height = int(width), int(height)
    if width % 8 or height % 8:
        raise gr.Error("Width and height must be divisible by 8")

    seed = int(seed)
    if seed < 0:
        seed = random.SystemRandom().randint(0, 2**31 - 1)

    return GenerationJob(
        prompt=prompt,
        negative_prompt=negative_prompt.strip(),
        width=width,
        height=height,
        steps=int(steps),
        cfg_scale=float(cfg_scale),
        seed=seed,
    )


def generate_placeholder(job):
    digest = hashlib.sha256(
        f"{job.prompt}:{job.seed}".encode("utf-8")
    ).digest()
    color = tuple(48 + value % 160 for value in digest[:3])

    image = Image.new("RGB", (job.width, job.height), color)
    draw = ImageDraw.Draw(image)
    draw.multiline_text(
        (24, 24),
        f"Prompt: {job.prompt[:80]}\nSeed: {job.seed}\nSteps: {job.steps}",
        fill="white",
        spacing=8,
    )
    return image


def run_workflow(prompt, negative_prompt, width, height, steps, cfg_scale, seed):
    job = build_job(
        prompt, negative_prompt, width, height, steps, cfg_scale, seed
    )
    image = generate_placeholder(job)
    metadata = json.dumps(asdict(job), ensure_ascii=False, indent=2)
    return image, metadata, job.seed


with gr.Blocks(title="Workflow Image UI") as demo:
    gr.Markdown("# Workflow Image UI")

    with gr.Row():
        with gr.Column(scale=2):
            prompt = gr.Textbox(label="Prompt", lines=3)
            negative_prompt = gr.Textbox(label="Negative prompt", lines=2)

            with gr.Row():
                width = gr.Slider(256, 1024, value=512, step=64, label="Width")
                height = gr.Slider(256, 1024, value=512, step=64, label="Height")

            steps = gr.Slider(1, 80, value=20, step=1, label="Steps")
            cfg_scale = gr.Slider(1, 20, value=7, step=0.5, label="CFG scale")
            seed = gr.Number(value=-1, precision=0, label="Seed (-1 for random)")
            generate = gr.Button("Generate", variant="primary")

        with gr.Column(scale=3):
            output = gr.Image(label="Result", type="pil")
            actual_seed = gr.Number(label="Used seed", precision=0)
            metadata = gr.Code(label="Generation metadata", language="json")

    generate.click(
        fn=run_workflow,
        inputs=[prompt, negative_prompt, width, height, steps, cfg_scale, seed],
        outputs=[output, metadata, actual_seed],
    )

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch()

运行：

python -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
python app.py

Windows PowerShell 中将激活命令改为：

.venv\Scripts\Activate.ps1

接入真实推理后端时，可以让 generate_placeholder 调用本地 pipeline，也可以提交 HTTP 任务：

import requests


def generate_remote(job):
    response = requests.post(
        "http://127.0.0.1:8000/v1/txt2img",
        json=asdict(job),
        timeout=300,
    )
    response.raise_for_status()
    return response.json()

生产环境中不要只设置一个很长的 HTTP 超时。更稳妥的方式是返回任务 ID，由前端轮询状态，并提供取消接口。

AUTOMATIC1111 式能力应该怎样逐步迁移

不要一开始就追求全部功能对齐。建议按依赖关系推进：

先打通 txt2img、固定模型和单图输出，并记录完整元数据。
加入模型与 VAE 选择，同时建立进程级缓存和显存回收策略。
扩展 img2img、蒙版和批量任务，明确文件生命周期。
再引入 LoRA、ControlNet 或高清修复等可选节点。
最后处理插件机制、权限隔离和多用户并发。

模型对象不应在每次点击时重新创建。可以按模型标识缓存 pipeline，但必须限制缓存数量，并在切换模型时正确释放 GPU 引用。多用户部署还需要检查提示词、上传文件、输出目录和插件代码的隔离边界。

上线前的检查清单

相同模型、参数和 seed 能否重放结果。
工作流节点是否记录耗时、异常和任务 ID。
用户取消任务后，队列槽位与显存是否真正释放。
上传图片与生成文件是否有大小、类型和保留期限限制。
模型切换是否会导致并发请求使用错误实例。
Gradio 是否只负责交互，而推理契约仍可被 API、脚本和测试复用。

用 Gradio Workflow 重建 AUTOMATIC1111 的关键，不是把旧界面换一种写法，而是把隐藏在界面回调里的推理过程变成清晰的数据流。界面可以快速迭代，模型后端也可以替换；稳定的数据契约、任务状态和资源治理，才决定这个重建项目能否从演示走向长期使用。