学术研究 | ComfyUI源码分析-NODE RUNTIME TRACE

class CLIPTextEncode:    @classmethod    def INPUT_TYPES(s):        return {"required": {"text": ("STRING", {"multiline": True}), "clip": ("CLIP", )}}    RETURN_TYPES = ("CONDITIONING",)    FUNCTION = "encode"
    CATEGORY = "conditioning"
    def encode(self, clip, text):        tokens = clip.tokenize(text)        cond, pooled = clip.encode_from_tokens(tokens, return_pooled=True)        return ([[cond, {"pooled_output": pooled}]], )

def init_custom_nodes():    extras_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy_extras")    extras_files = [        "nodes_latent.py",        "nodes_hypernetwork.py",

for node_file in extras_files:        load_custom_node(os.path.join(extras_dir, node_file)) # 没有传ignore

# 比如：comfy_extras/nodes_tomesd.pyNODE_CLASS_MAPPINGS = {    "TomePatchModel": TomePatchModel,}

def load_custom_nodes():    base_node_names = set(NODE_CLASS_MAPPINGS.keys())    node_paths = folder_paths.get_folder_paths("custom_nodes")    for custom_node_path in node_paths:        possible_modules = os.listdir(custom_node_path)        for possible_module in possible_modules:            module_path = os.path.join(custom_node_path, possible_module)            if os.path.isfile(module_path) and os.path.splitext(module_path)[1] != ".py": continue            if module_path.endswith(".disabled"): continue            success = load_custom_node(module_path, base_node_names) # ignore=base_node_names

unique_id   class_type7           CLIPTextEncode5           EmptyLatentImage3           KSampler8           VAEDecode4           CheckpointLoaderSimple10          PreviewImage6           CLIPTextEncode

结果节点: [10]node列表：{[3, info3], [4, info4], [5, info5], [6, info6], [7, info7], [8, info8], [9, info9], [10, info10]}

def recursive_will_execute(prompt, outputs, current_item):    """    递归获取将要执行的节点列表    """    # 获取当前节点的唯一标识符    unique_id = current_item    # 获取当前节点的输入数据    inputs = prompt[unique_id]['inputs']     # 存储将要执行的节点列表    will_execute = []    # 如果当前节点已经在输出中，则直接返回空列表    if unique_id in outputs:        return []
    # 遍历当前节点的输入    for x in inputs:        input_data = inputs[x]        # 如果输入数据是一个列表        if isinstance(input_data, list):            # 获取输入数据的唯一标识符和输出索引            input_unique_id = input_data[0]            output_index = input_data[1]            # 如果输入数据的唯一标识符不在输出中，则递归获取将要执行的节点列表            if input_unique_id not in outputs:                will_execute += recursive_will_execute(prompt, outputs, input_unique_id)
    # 返回将要执行的节点列表，包括当前节点的唯一标识符    return will_execute + [unique_id]

def get_input_data(inputs, class_def, unique_id, outputs={}, prompt={}, extra_data={}):    """    该函数的主要功能是根据输入参数 inputs、class_def、unique_id, 以及可选参数 outputs、prompt、extra_data, 生成一个包含所有输入数据的字典 input_data_all。    """    valid_inputs = class_def.INPUT_TYPES()    input_data_all = {}    for x in inputs:        # 一部分数据来自输出        input_data = inputs[x]        if isinstance(input_data, list):            input_unique_id = input_data[0]            output_index = input_data[1]            if input_unique_id not in outputs:                input_data_all[x] = (None,)                continue            obj = outputs[input_unique_id][output_index]            input_data_all[x] = obj        else:            # 一部分数据来自输入            if ("required" in valid_inputs and x in valid_inputs["required"]) or ("optional" in valid_inputs and x in valid_inputs["optional"]):                input_data_all[x] = [input_data]

# get a slice of inputs, repeat last input when list isn't long enough    def slice_dict(d, i):        d_new = dict()        for k,v in d.items():            d_new[k] = v[i if len(v) > i else -1]        return d_new        results = []    if input_is_list:        if allow_interrupt:            nodes.before_node_execution()        results.append(getattr(obj, func)(**input_data_all))    elif max_len_input == 0:        if allow_interrupt:            nodes.before_node_execution()        results.append(getattr(obj, func)())    else:        for i in range(max_len_input):            if allow_interrupt:                nodes.before_node_execution()            results.append(getattr(obj, func)(**slice_dict(input_data_all, i)))    return results

1.每个node的类型都要已注册（在`NODE_CLASS_MAPPINGS`中可查）。2.每个请求的workflow至少要有一个输出节点（node的定义含有`OUTPUT_NODE`属性并且`node.OUTPUT_NODE == True`）。3.判断node输入定义`INPUT_TYPES`中的必填字段（`required`相关字段）有没有在输入里面。4.判断依赖node链接是否正常。首先判断依赖“坐标”（前面提到的`m_n`，比如10依赖8_0）node是否真实存在、类型是否正确，然后本节点的输入类型和依赖node的对应位输出类型（`RETURN_TYPES[n]`）是否一致。5.基础类型校验。如果是`INT`、`FLOAT`、`STRING`这些基础类型，会进行类型强转判断，如果有设置`min`、`max`也会判断最小值和最大值范围。6.自定义输入校验。ComfyUI的链路校验相对较弱，如果node需要自己的特殊校验，可以实现自定义校验函数`VALIDATE_INPUTS`。

self.diffusion_model = UNetModel(**unet_config, device=device) def load_model_weights(self, sd, unet_prefix="model.diffusion_model."):        to_load = {}        keys = list(sd.keys())        for k in keys:            if k.startswith(unet_prefix):                to_load[k[len(unet_prefix):]] = sd.pop(k)
        m, u = self.diffusion_model.load_state_dict(to_load, strict=False)

vae_sd = comfy.utils.state_dict_prefix_replace(sd, {"first_stage_model.": ""}, filter_keys=True)vae_sd = model_config.process_vae_state_dict(vae_sd)vae = VAE(sd=vae_sd)...self.first_stage_model = AutoencoderKL(ddconfig=ddconfig, embed_dim=4)m, u = self.first_stage_model.load_state_dict(sd, strict=False)

def clip_target(self):        return supported_models_base.ClipTarget(sd1_clip.SD1Tokenizer, sd1_clip.SD1ClipModel)

text='beautiful girl smile'。# 1 分词tokens = clip.tokenize(text) # text_model的输入token是77个（算上开始和结束符），超过截断，不够‘补0’。# 即：[<start>,<beautiful>,<girl>,<smile>,<end>,...,<end>]共`77`个，下面的1.0表示词权。# tokens = [(49406,1.0),(1215,1.0)(1611,1.0)(3490,1.0)(49407,1.0)...(49407,1.0)]# 2 load model to gpu# 3 text model encoder，获取构造onditioningcond, pooled = clip.encode_from_tokens(tokens, return_pooled=True)# 每个token生成大小为768的embedding，所以cond的shape是`torch.Size([1, 77, 768])`。

# 即长宽减少8倍，通道扩成4.# 这里有个点需要注意，图片的长宽最好是8的倍数。latent = torch.zeros([batch_size, 4, height // 8, width // 8])

旧格式：positive: list (1) [pub]| [0]: list (2) [pub]| | [0]: Tensor (1)| | [1]: dict (1) [pub]| | | ['pooled_output']: Tensor (1)---------# 调用栈3 格式转换，`model_conds["c_crossattn"] = comfy.conds.CONDCrossAttn(c[0])`positive = convert_cond(positive)negative = convert_cond(negative)--------新格式：positive: list (1) [pub]| [0]: dict (2) [pub]| | ['pooled_output']: Tensor (1)| | ['model_conds']: dict (1) [pub]| | | ['c_crossattn']: CONDCrossAttn
---------# 调用栈11 格式转换，positive和negative的c_crossattn做cond_cat， `c = cond_cat(c)`---------新格式：c: dict (2) [pub]| ['c_crossattn']: Tensor (2) [pub]| | [0]: Tensor (77)| | [1]: Tensor (77)最终unet收到的是：context=c['c_crossattn']

# 调用栈2 在T2I场景，latent_image的shape会影响noise的生成。noise = comfy.sample.prepare_noise(latent_image, seed, batch_inds)
# 调用栈5，`latent_image = model.process_latent_in(latent_image)`。latent_image乘系数，该步对T2I没影响，latent_image本身就是0。`return latent * self.scale_factor`。
# 调用栈6，`noise += latent_image`, 噪声叠加latent_image信息，对I2I有影响，对T2I没影响。
# 调用栈8，I2I mask场景，影响输出。if denoise_mask is not None:            out += self.latent_image * latent_mask

# 调用栈2，seed影响noise创建。`generator = torch.manual_seed(seed)`

[14.6146, 11.9175,  9.8142,  8.1584,  6.8430,  5.7885,  4.9356,  4.2397,  3.6669,  3.1913,  2.7931,  2.4569,  2.1705,  1.9246,  1.7116,  1.5257,  1.3619,  1.2166,  1.0865,  0.9691,  0.8622,  0.7640,  0.6730,  0.5877,  0.5067,  0.4286,  0.3515,  0.2722,  0.1835,  0.0292,  0.0000]

# 调用栈3，# denoise 会影响 stepdef set_steps(self, steps, denoise=None):        self.steps = steps        if denoise is None or denoise > 0.9999:            self.sigmas = self.calculate_sigmas(steps).to(self.device)        else:            new_steps = int(steps/denoise)            sigmas = self.calculate_sigmas(new_steps).to(self.device)            self.sigmas = sigmas[-(steps + 1):]
# scheduler_name steps会影响sigmasdef calculate_sigmas_scheduler(model, scheduler_name, steps):    if scheduler_name == "karras":        sigmas = k_diffusion_sampling.get_sigmas_karras(n=steps, sigma_min=float(model.model_sampling.sigma_min), sigma_max=float(model.model_sampling.sigma_max))    elif scheduler_name == "exponential":        sigmas = k_diffusion_sampling.get_sigmas_exponential(n=steps, sigma_min=float(model.model_sampling.sigma_min), sigma_max=float(model.model_sampling.sigma_max))    elif scheduler_name == "normal":        sigmas = normal_scheduler(model, steps)    elif scheduler_name == "simple":        sigmas = simple_scheduler(model, steps)    elif scheduler_name == "ddim_uniform":        sigmas = ddim_scheduler(model, steps)    elif scheduler_name == "sgm_uniform":        sigmas = normal_scheduler(model, steps, sgm=True)    else:        print("error invalid scheduler", self.scheduler)    return sigmas

# 调用栈11，这里要注意，这里的代码是拿到unet处理结果的后置处理。# 这里的 cond_scale=cfgreturn uncond + (cond - uncond) * cond_scale

KSAMPLER_NAMES = ["euler", "euler_ancestral", "heun", "heunpp2","dpm_2", "dpm_2_ancestral",                  "lms", "dpm_fast", "dpm_adaptive", "dpmpp_2s_ancestral", "dpmpp_sde", "dpmpp_sde_gpu",                  "dpmpp_2m", "dpmpp_2m_sde", "dpmpp_2m_sde_gpu", "dpmpp_3m_sde", "dpmpp_3m_sde_gpu", "ddpm", "lcm"]

#a111 v1.5.0 webui梳理核心的python文件：processing.py   - 流程处理：   process_images_inner   https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L673   https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L689   - 核心的两个类：             - StableDiffusionProcessingTxt2Img       https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L922       核心的函数处理：def sample(       https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L1034             - StableDiffusionProcessingImg2Img                https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L1210                核心的函数处理：def sample(                https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/processing.py#L1363核心的python文件：sd_samplers_kdiffusion.py，采样器及步数迭代       - 采样器选择：https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/sd_samplers_kdiffusion.py#L13       - 处理 text2image : https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/sd_samplers_kdiffusion.py#L441       - 处理 image2image : https://github.com/AUTOMATIC1111/stable-diffusion-webui/blob/v1.5.0/modules/sd_samplers_kdiffusion.py#L397

# -*- coding: utf-8 -*-
class Input:    @classmethod    def INPUT_TYPES(s):        return {"required": {"number": ("FLOAT", {"multiline": True})}}    RETURN_TYPES = ("CalcFLOAT",)    FUNCTION = "input"    CATEGORY = "calc"
    def input(self, number):        return (number, )    class Add:    @classmethod    def INPUT_TYPES(s):        return {"required": {"number1": ("CalcFLOAT",), "number2": ("CalcFLOAT",)}}    RETURN_TYPES = ("CalcSTR",)    FUNCTION = "add"    CATEGORY = "calc"
    def add(self, number1, number2):        val = str(number1+number2)        return (val, )
NODE_CLASS_MAPPINGS = {    "Input": Input,    "Add": Add,}