ComfyUI学习笔记（一）：环境搭建

1.创建基础环境

1.1 拉取基础镜像

>> docker pull nvidia/cuda:12.4.1-devel-ubuntu22.04

1.2 创建容器环境

>> docker run -itd --name mirror --shm-size 32g \
 --ipc=host --net=host --privileged=true \
 --cap-add=SYS_PTRACE --workdir /root/workspace \
 -v /data/project/jingyu/workspace:/root/workspace \
 --gpus all nvidia/cuda:12.4.1-devel-ubuntu22.04 bash

1.3 安装基础工具

# 1.进入容器
>> docker exec -it mirror bash

# 2.换软件源
>> sed -i 's|http://.*.ubuntu.com|http://mirrors.tuna.tsinghua.edu.cn6|g' \
/etc/apt/sources.list
>> sed -i 's/mirrors.tuna.tsinghua.edu.cn6/mirrors.tuna.tsinghua.edu.cn/g' \
/etc/apt/sources.list

# 3.安装基础组件
>> apt update && apt-get install -y gcc g++ make cmake zlib1g zlib1g-dev \
openssl libsqlite3-dev libssl-dev libffi-dev libbz2-dev libxslt1-dev unzip \
pciutils net-tools libblas-dev gfortran libblas3 vim zip wget git \
build-essential kmod pciutils systemd ffmpeg libavcodec-dev libavformat-dev \
libavutil-dev libswscale-dev libavfilter-dev libavdevice-dev \
libx264-dev libxvidcore-dev curl

# 4.添加dns解析
>> echo "nameserver 101.6.6.6" > /etc/resolv.conf && echo "nameserver 223.5.5.5" >> /etc/resolv.conf

# 5.安装zsh
>> apt install zsh
## （1）能访问github
>> sh -c "$(wget -O- https://raw.githubusercontent.com/ohmyzsh/ohmyzsh/master/tools/install.sh)"
## （2）不能访问github
>> git clone https://gitee.com/mirrors/oh-my-zsh.git ~/.oh-my-zsh
>> cp ~/.oh-my-zsh/templates/zshrc.zsh-template ~/.zshrc
## （3）刷新配置
>> zsh && vim ~/.zshrc
## 5.4 配置插件
plugins=(git z zsh-autosuggestions extract web-search zsh-syntax-highlighting colored-man-pages history)
## 5.5 安装插件
# （1）github
>> git clone https://github.com/zsh-users/zsh-autosuggestions ~/.oh-my-zsh/custom/plugins/zsh-autosuggestions
>> git clone https://github.com/zsh-users/zsh-syntax-highlighting.git ~/.oh-my-zsh/custom/plugins/zsh-syntax-highlighting
# （2）gitee
>> git clone https://gitee.com/sasukeZhou/zsh-autosuggestions.git ~/.oh-my-zsh/custom/plugins/zsh-autosuggestions
>> git clone https://gitee.com/pankla/zsh-syntax-highlighting.git ~/.oh-my-zsh/custom/plugins/zsh-syntax-highlighting
>> source ~/.zshrc

# 6.安装conda
>> cd && mkdir software && cd software 
>> wget -4 https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh --no-check-certificate
>> bash ./Miniconda3-latest-Linux-x86_64.sh
>> source ~/.zshrc

2.安装pytorch环境

 不同版本的Pytorch安装命令可以从Pytorch安装命令查询得到，我这里安装2.6.0

 安装pytorch的preview版本，默认会安装triton

# 1.创建开发环境
>> conda create -n comfyui python=3.12 

# 2.切换开发环境
>> vim ~/.zshrc
conda activate comfyui
>> source ~/.zshrc

# 3.安装pre版本pytorch
>> pip install --pre torch==2.6.0 torchvision==0.21.0 torchaudio==2.6.0 \
--index-url https://download.pytorch.org/whl/cu124

# 4.查看是否安装成功
>> python
Python 3.12.12 | packaged by Anaconda, Inc. | (main, Oct 21 2025, 20:16:04) [GCC 11.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import torch
>>> torch.cuda.is_available()
True
>>> import triton
>>> print(f"Triton 已安装，版本号：{triton.__version__}")
Triton 已安装，版本号：3.2.0
>>> exit()

# 5.安装nunchaku插件：https://github.com/nunchaku-ai/nunchaku/releases
>> pip install nunchaku-1.0.1+torch2.6-cp312-cp312-linux_x86_64.whl

 最近感觉国内源好像没有官方源快，就直接用官方源吧。

3.安装ComfyUI

# 1.下载源码：默认已装git，由于cuda版本较旧，
# 最新版本需要12.8及以上才能玩，所以这里用v0.7.0
>> git clone -b v0.7.0 https://github.com/Comfy-Org/ComfyUI.git
# 2.安装ComfyUI环境依赖
>> cd ComfyUI && pip install -r requirements.txt
# 3.安装ComfyUI-Manager环境依赖
>> pip install -r manager_requirements.txt

4.启动ComfyUI服务

• （1）启动服务：

# 1.直接启动
>> python ./main.py --enable-manager --port 10086 --listen 0.0.0.0
Total VRAM 22906 MB, total RAM 257566 MB
pytorch version: 2.6.0+cu124
Set vram state to: NORMAL_VRAM
Device: cuda:0 Tesla P40 : cudaMallocAsync
Using async weight offloading with 2 streams
Enabled pinned memory 244687.0

Import times for custom nodes:
   0.0 seconds: /home/mirror/workspace/ComfyUI/custom_nodes/websocket_image_save.py

Context impl SQLiteImpl.
Will assume non-transactional DDL.
No target revision found.
Starting server

To see the GUI go to: http://0.0.0.0:10086

# 2.后台启动，并指定显卡
>> CUDA_VISIBLE_DEVICES=3 nohup python main.py --listen 0.0.0.0 --port 10086 > comfyui.log 2>&1 &

• （2）访问服务：网页端打开<服务器ip>:10086即可：
  

5.文生图工作流

5.1 打开Z-Image-Turbo文生图工作流：

5.2 下载所需模型

• （1）models/text_encoders/qwen_3_4b.safetensors：8.04GB
• （2）models/vae/ae.safetensors：335.30MB
• （3）models/diffusion_models/z_image_turbo_bf16.safetensors：12.31GB
• （4）models/loras/pixel_art_style_z_image_turbo.safetensors：170.13MB

5.3 直接执行工作流

 调整一下提示词后运行，提示词调整为：生成一个红色的苹果，局部出现腐烂

5.4 服务请求执行工作流

 先看一下文档关于核心API路由介绍：ComfyUI-Server

路径	get/post/ws	用途
/	get	加载 Comfy 网页
/ws	websocket	用于与服务器进行实时通信的 WebSocket 端点
/embeddings	get	获取可用的嵌入模型名称列表
/extensions	get	获取注册了 WEB_DIRECTORY 的扩展列表
/features	get	获取服务器功能和能力
/models	get	获取可用模型类型列表
/models/{folder}	get	获取特定文件夹中的模型
/workflow_templates	get	获取自定义节点模块及其关联模板工作流的映射
/upload/image	post	上传图片
/upload/mask	post	上传蒙版
/view	get	查看图片。更多选项请参见 server.py 中的 @routes.get("/view")
/view_metadata/	get	获取模型的元数据
/system_stats	get	获取系统信息（Python 版本、设备、显存等）
/prompt	get	获取当前队列状态和执行信息
/prompt	post	提交提示到队列
/object_info	get	获取所有节点类型的详细信息
/object_info/{node_class}	get	获取特定节点类型的详细信息
/history	get	获取队列历史记录
/history/{prompt_id}	get	获取特定提示的队列历史记录
/history	post	清除历史记录或删除历史记录项
/queue	get	获取执行队列的当前状态
/queue	post	管理队列操作（清除待处理/运行中的任务）
/interrupt	post	停止当前工作流执行
/free	post	通过卸载指定模型释放内存
/userdata	get	列出指定目录中的用户数据文件
/v2/userdata	get	增强版本，以结构化格式列出文件和目录
/userdata/{file}	get	获取特定的用户数据文件
/userdata/{file}	post	上传或更新用户数据文件
/userdata/{file}	delete	删除特定的用户数据文件
/userdata/{file}/move/{dest}	post	移动或重命名用户数据文件
/users	get	获取用户信息
/users	post	创建新用户（仅限多用户模式）

 对于文生图任务而言，需要三步来实现工作流调用：(1) 调用/prompt将工作流提交到队列；(2) 调用/history{prompt_id}获取当前任务的队列记录；(3) 调用/view获取对应结果。

 具体操作步骤：

• （1）导出对应API配置：

• （2）请求服务：

'''
Author: chenjingyu
Date: 2025-11-26 15:33:14
Contact: 2458006466@qq.com
Description: z_image
'''
import sys
import json
import time
import uuid
import requests
from urllib.parse import urlencode
from pathlib import Path

curr_path = Path(__file__).parent.resolve()
if str(curr_path) not in sys.path:
    sys.path.append(str(curr_path))
    
req_url = 'http://修改为你的服务器IP'
req_port = 10086

# 1.载入工作流
def load_config(path: str):
    with open(path, 'r', encoding='utf-8') as fp:
        return json.load(fp)

# 2.修改对应提示词内容，并调用/prompt接口提交任务
def submit_prompt(url: str, port: int, wf: dict):         
    payload = {"prompt": wf, "client_id": str(uuid.uuid4())}
    resp = requests.post(f"{url}:{port}/prompt", json=payload).json()
    print(resp)
    if "error" in resp:
        raise RuntimeError(resp["error"])
    return resp["prompt_id"]

# 3.调用/history接口查看当前任务执行情况，执行完毕后，调用/view接口获取生成内容
def wait_and_download(url: str, port: int, prompt_id: str, save_dir: str):
    while True:
        time.sleep(1)
        hist = requests.get(f"{url}:{port}/history/{prompt_id}").json()
        if prompt_id not in hist:
            continue
        for _, outs in hist[prompt_id]["outputs"].items():
            if "images" in outs:
                for idx, element in enumerate(outs["images"]):
                    url = f"{url}:{port}/view?{urlencode(element)}"
                    save_name = f"{save_dir}/{idx}.png"
                    Path(save_name).write_bytes(requests.get(url).content)
                    print(f"保存  {save_name}")
                return                                  # 全部下完就结束
        break
    raise FileNotFoundError("未找到输出结果")

# 主调用流程 
def text_to_image_z_image(url: str, port: int, wf_path: str, node_name: str, prompt: str, save_path: str):
    wf = load_config(wf_path)
    wf[node_name]["inputs"]["text"] = prompt
    if not Path(save_path).exists():
        Path(save_path).mkdir(exist_ok=True, parents=True)
    try:
        start = time.time()
        pid = submit_prompt(url, port, wf)
        print(f"已提交，prompt_id={pid}")
        wait_and_download(url, port, pid, save_path)
        end = time.time()
        print(f"总用时：{end-start:.2f}s")
        return True
    except Exception as e:
        print(f"Error: {e}")
        return False

if __name__ == '__main__':
    in_node_name = "45"
    wf_path = f"{curr_path}/image_z_image_turbo.json"
    text_to_image_z_image(req_url, req_port, wf_path, in_node_name, '生成一张桌子，上面摆放一个白色的花瓶', './output')

• （3）执行任务：

>> python z_image.py
{'prompt_id': '2e2ce1d0-e57d-498a-acb0-2abe1a9813ca', 'number': 5, 'node_errors': {}}
已提交，prompt_id=2e2ce1d0-e57d-498a-acb0-2abe1a9813ca
保存  ./output/0.png
总用时：8.93s

• （4）生成结果：

5.5 一些API请求的辅助函数

'''
Author: chenjingyu
Date: 2026-01-20 15:31:18
Contact: 2458006466@qq.com
Description: utils
'''
import sys
import uuid
import time
import json
import requests
from pathlib import Path
from urllib.parse import urlencode

curr_path = Path(__file__).parent.resolve()
if str(curr_path) not in sys.path:
    sys.path.append(str(curr_path))


# 1.载入工作流
def load_config(path: str):
    with open(path, 'r', encoding='utf-8') as fp:
        return json.load(fp)


# 2.上传图像
def upload_image(req_url: str, path: str):
    with open(path, "rb") as f:
        r = requests.post(f"{req_url}/upload/image", files={"image": f})
    r.raise_for_status()
    return r.json()["name"]


# 3.提交任务
def submit_prompt(req_url: str, wf: dict):         
    payload = {"prompt": wf, "client_id": str(uuid.uuid4())}
    resp = requests.post(f"{req_url}/prompt", json=payload).json()
    print(resp)
    if "error" in resp:
        raise RuntimeError(resp["error"])
    return resp["prompt_id"]


# 4.获取当前任务保存结果后缀
def get_prefix(wf: dict, output_node_name: str):
    class_type = wf[output_node_name]["class_type"]
    if class_type == "SaveVideo":
        return 'mp4'
    elif class_type == "SaveImage":
        return "png"
    else:
        raise NotImplementedError(f"不支持当前类型: {class_type}")
    

# 5.轮询直到完成
def wait_and_download(req_url: str, prompt_id: str, output_node_name: str, save_dir: str, prefix: str):
    while True:
        time.sleep(1)
        hist = requests.get(f"{req_url}/history/{prompt_id}").json()
        if prompt_id not in hist:
            continue
        save_path = Path(save_dir)
        if not Path.exists(save_path):
            Path.mkdir(save_path, parents=True, exist_ok=True)
        prompt_data = hist[prompt_id]
        for node_name, outs in prompt_data["outputs"].items():
            if node_name == output_node_name and "images" in outs:
                for idx, item in enumerate(outs["images"]):
                    url =  f"{req_url}/view?{urlencode(item)}"
                    save_name = f"{save_path}/{idx}.{prefix}"
                    Path(save_name).write_bytes(requests.get(url).content)
                    print(f"保存  {save_name}")
                return
        break
    raise FileNotFoundError("未找到输出结果")

# 6.获取队列状态
def get_queue_status(req_url: str):
    try:
        resp = requests.get(f"{req_url}/queue")
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        print(f"查询任务队列状态失败：{e}")
        return None
      
if __name__ == "__main__":
    req_url = "http://<服务地址>:<服务IP>"
    queue_status = get_queue_status(req_url)
    if queue_status.get("queue_running"):
        print("service is busy.")
        print(queue_status["queue_running"])

参考资料

• [1] ComfyUI
• [2] ComfyUI-Server