拆解 SD 潜空间 Img2Img：从零搭建可 API 调度的 ComfyUI 底层绘图流水线

CheckpointLoaderSimple（加载SD大模型，输出 MODEL / CLIP / VAE）
├─ CLIPTextEncode（正向提示词，CLIP输入 → CONDITION）
├─ CLIPTextEncode（反向提示词，CLIP输入 → CONDITION）
└─ VAE → VAEEncode 的 vae输入槽
LoadImage（上传参考原图，image输出）
└─ VAEEncode（image输入，vae来自Checkpoint；输出 latent_image 给KSampler）
KSampler
├─ model：Checkpoint MODEL
├─ positive：正向CLIP条件
├─ negative：反向CLIP条件
├─ latent_image：VAEEncode输出潜图
├─ seed、steps、cfg、sampler_name、scheduler、denoising_strength 参数组
└─ latent输出 → VAEDecode
VAEDecode
├─ samples：KSampler潜输出
├─ vae：Checkpoint VAE
└─ image输出 → SaveImage 保存节点

{
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 12345,
      "steps": 25,
      "cfg": 7.0,
      "sampler_name": "dpmpp_2m",
      "scheduler": "karras",
      "denoising_strength": 0.5,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["10", 0]
    }
  },
  "4": {
    "class_type": "CheckpointLoaderSimple",
    "inputs": {
      "ckpt_name": "v1-5-pruned-emaonly.safetensors"
    }
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "masterpiece, best quality, anime style, beautiful girl, soft lighting",
      "clip": ["4", 1]
    }
  },
  "7": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "worst quality, low quality, blurry, deformed, extra limbs, watermark",
      "clip": ["4", 1]
    }
  },
  "10": {
    "class_type": "VAEEncode",
    "inputs": {
      "pixels": ["11", 0],
      "vae": ["4", 2]
    }
  },
  "11": {
    "class_type": "LoadImage",
    "inputs": {
      "image": "input_ref.png"
    }
  },
  "12": {
    "class_type": "VAEDecode",
    "inputs": {
      "samples": ["3", 0],
      "vae": ["4", 2]
    }
  },
  "13": {
    "class_type": "SaveImage",
    "inputs": {
      "images": ["12", 0],
      "filename_prefix": "img2img_out"
    }
  }
}

pip install requests pillow

import requests
import json
import os
import time
from PIL import Image
from io import BytesIO
COMFY_API_URL = "http://127.0.0.1:8188"
INPUT_DIR = "./input"
OUTPUT_DIR = "./output"
os.makedirs(INPUT_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
# 基础工作流模板，和上面API JSON对应
BASE_WORKFLOW = {
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 12345,
      "steps": 25,
      "cfg": 7.0,
      "sampler_name": "dpmpp_2m",
      "scheduler": "karras",
      "denoising_strength": 0.5,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["10", 0]
    }
  },
  "4": {
    "class_type": "CheckpointLoaderSimple",
    "inputs": {
      "ckpt_name": "v1-5-pruned-emaonly.safetensors"
    }
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "masterpiece, best quality, anime, soft color, detailed face",
      "clip": ["4", 1]
    }
  },
  "7": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "worst quality, lowres, blurry, bad anatomy, ugly, watermark",
      "clip": ["4", 1]
    }
  },
  "10": {
    "class_type": "VAEEncode",
    "inputs": {
      "pixels": ["11", 0],
      "vae": ["4", 2]
    }
  },
  "11": {
    "class_type": "LoadImage",
    "inputs": {
      "image": ""
    }
  },
  "12": {
    "class_type": "VAEDecode",
    "inputs": {
      "samples": ["3", 0],
      "vae": ["4", 2]
    }
  },
  "13": {
    "class_type": "SaveImage",
    "inputs": {
      "images": ["12", 0],
      "filename_prefix": "img2img_batch"
    }
  }
}
def upload_image(file_path):
    """上传本地图片到ComfyUI服务端，返回服务端内文件名"""
    with open(file_path, "rb") as f:
        res = requests.post(f"{COMFY_API_URL}/upload/image", files={"image": f})
    return res.json()["name"]
def submit_workflow(workflow):
    payload = {"prompt": workflow}
    r = requests.post(f"{COMFY_API_URL}/prompt", json=payload)
    return r.json()["prompt_id"]
def wait_for_result(prompt_id):
    while True:
        history = requests.get(f"{COMFY_API_URL}/history").json()
        if prompt_id in history:
            return history[prompt_id]
        time.sleep(0.5)
def save_output(history_data, out_name):
    outputs = history_data["outputs"]
    for node_id, node_out in outputs.items():
        if "images" in node_out:
            for img_info in node_out["images"]:
                img_data = requests.get(f"{COMFY_API_URL}/view", params=img_info).content
                with open(os.path.join(OUTPUT_DIR, out_name), "wb") as f:
                    f.write(img_data)
                print(f"✅ 已保存：{out_name}")
def batch_run(denoise=0.5):
    img_list = [f for f in os.listdir(INPUT_DIR) if f.lower().endswith(("png","jpg","jpeg"))]
    for fname in img_list:
        full_path = os.path.join(INPUT_DIR, fname)
        print(f"\n🚀 处理图片：{fname}")
        try:
            server_img_name = upload_image(full_path)
            wf = json.loads(json.dumps(BASE_WORKFLOW))
            wf["3"]["inputs"]["denoising_strength"] = denoise
            wf["11"]["inputs"]["image"] = server_img_name
            pid = submit_workflow(wf)
            res_data = wait_for_result(pid)
            save_output(res_data, f"out_{fname}")
        except Exception as e:
            print(f"❌ {fname} 处理失败：{str(e)}")
if __name__ == "__main__":
    # 修改denoise参数适配你的业务
    batch_run(denoise=0.5)