migrate to v3 schema and add LoRA extraction node

.gitignore

@@ -1,3 +1,7 @@
+# Repo specific
+AGENTS.MD
+custom_nodes.examples.md
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

__init__.py

@@ -1,17 +1,52 @@
-from .nodes import *
+from typing_extensions import override
+from comfy_api.latest import ComfyExtension, io
 
-NODE_CLASS_MAPPINGS = {
-    "UNetMetaKeys": UNetMetaKeys,
-    "CLIPMetaKeys": CLIPMetaKeys,
-    "LoRAMetaKeys": LoRAMetaKeys,
-    "CheckpointMetaKeys": CheckpointMetaKeys,
-    "OffloadModels": OffloadModelsNode,
-}
+from .nodes.metakeys import (
+    ModelMetaKeys, TextEncoderMetaKeys, LoRAMetaKeys,
+    CheckpointMetaKeys, EmbeddingMetaKeys
+)
+from .nodes.renamekeys import (
+    ModelRenameKeys, TextEncoderRenameKeys, LoRARenameKeys,
+    CheckpointRenameKeys, EmbeddingRenameKeys
+)
+from .nodes.prunekeys import (
+    ModelPruneKeys, TextEncoderPruneKeys, LoRAPruneKeys,
+    CheckpointPruneKeys, EmbeddingPruneKeys
+)
+from .nodes.merger import (
+    ModelTwoMerger, TextEncoderTwoMerger, LoRATwoMerger,
+    CheckpointTwoMerger, EmbeddingTwoMerger,
+    ModelThreeMerger, TextEncoderThreeMerger, LoRAThreeMerger,
+    CheckpointThreeMerger, EmbeddingThreeMerger
+)
+from .nodes.lora_extract import (
+    LoRASaveFromFile, LoRACheckpointSaveFromFile
+)
 
-NODE_DISPLAY_NAME_MAPPINGS = {
-    "UNetMetaKeys": "Get UNet Metadata and Keys",
-    "CLIPMetaKeys": "Get CLIP Metadata and Keys",
-    "LoRAMetaKeys": "Get LoRA Metadata and Keys",
-    "CheckpointMetaKeys": "Get Checkpoint Metadata and Keys",
-    "OffloadModelsNode": "Offload Models with Passthrough",
-}
+
+class ModelUtilsExtension(ComfyExtension):
+    @override
+    async def get_node_list(self) -> list[type[io.ComfyNode]]:
+        return [
+            # MetaKeys
+            ModelMetaKeys, TextEncoderMetaKeys, LoRAMetaKeys,
+            CheckpointMetaKeys, EmbeddingMetaKeys,
+            # RenameKeys
+            ModelRenameKeys, TextEncoderRenameKeys, LoRARenameKeys,
+            CheckpointRenameKeys, EmbeddingRenameKeys,
+            # PruneKeys
+            ModelPruneKeys, TextEncoderPruneKeys, LoRAPruneKeys,
+            CheckpointPruneKeys, EmbeddingPruneKeys,
+            # Two-Model Mergers
+            ModelTwoMerger, TextEncoderTwoMerger, LoRATwoMerger,
+            CheckpointTwoMerger, EmbeddingTwoMerger,
+            # Three-Model Mergers
+            ModelThreeMerger, TextEncoderThreeMerger, LoRAThreeMerger,
+            CheckpointThreeMerger, EmbeddingThreeMerger,
+            # LoRA Extraction
+            LoRASaveFromFile, LoRACheckpointSaveFromFile,
+        ]
+
+
+async def comfy_entrypoint() -> ModelUtilsExtension:
+    return ModelUtilsExtension()

docs/merger_2_model_modes.md

@@ -0,0 +1,87 @@
+# 2-Model Merger Documentation
+
+---
+
+## Weight-Sum
+> `A * (1 - α) + B * α`. A simple linear interpolation between two models.
+
+**Models Used:** A, B
+**Parameters:**
+- **Alpha:** Interpolation weight. `0.0` is 100% Model A, `1.0` is 100% Model B.
+
+---
+
+## Comparative-Interpolation
+> Interpolates between A and B based on the relative differences in their tensor values. Creates a unique blend by deciding how much of each model to use for each individual weight.
+
+**Models Used:** A, B
+**Parameters:**
+- **Alpha:** Controls the interpolation curve's strength and shape.
+- **Beta:** Switches the interpolation style. `0.0` focuses on similarity, `1.0` focuses on difference.
+- **Gamma:** Mixes between a randomized (binomial) interpolation at `0.0` and a smooth (linear) one at `1.0`.
+
+---
+
+## Power-Up (DARE)
+> Adds the unique capabilities of Model B to Model A using the Drop and Rescale (DARE) technique, which often preserves the knowledge of the base model better than simple additions.
+
+**Models Used:** A, B
+**Parameters:**
+- **Alpha:** The dropout rate. This is the proportion of unique weights from Model B that are *dropped* before merging. A higher value means less of B is merged.
+- **Beta:** A final multiplier for the rescaled difference before it's added to Model A.
+
+---
+
+## SVD LoRA Extraction
+> Extracts the difference between Model B and Model A and saves it as a new LoRA file. **This mode outputs a LoRA, not a full model.** The output file will be saved to your `loras` directory.
+
+**Models Used:** A (Tuned Model), B (Base Model)
+**Parameters:**
+- **Alpha:** The Rank (dimension) for the LoRA's standard layers.
+- **Beta:** The Rank (dimension) specifically for 3x3 convolution layers.
+- **Gamma:** The clamp quantile for weight values. `0.99` is a typical value to prevent outlier weights from dominating.
+
+---
+
+## Layer Mismatch Handling
+
+When merging models with different layer structures (e.g., fine-tuned models with added/removed layers, or LoRAs with partial layer coverage), the `mismatch_mode` parameter controls behavior:
+
+| Mode | Behavior |
+|------|----------|
+| `skip` | Missing layers in B use A's values **(default)** |
+| `zeros` | Missing layers in B are treated as zeros |
+| `error` | Fail if any layer is missing |
+
+**Common scenarios:**
+- Merging a fine-tuned model with added/removed layers
+- Combining LoRA-extracted differences with base models
+- Cross-architecture experiments
+
+**Note:** Extra layers in Model B that don't exist in Model A are currently ignored. The output will always have the same layer structure as Model A.
+
+---
+
+## Layer Filtering (Regex Patterns)
+
+Use regex patterns to control which layers are merged:
+
+### `exclude_patterns`
+Layers matching any pattern will **keep Model A's values only** (no merge).
+
+### `discard_patterns`
+Layers matching any pattern will be **removed entirely** from the output.
+
+**Pattern format:**
+- Whitespace-separated regex patterns
+- Patterns use **substring matching** (not full match)
+- Example: `text_model lora` matches any key containing "text_model" OR "lora"
+- Example: `layer\.[0-5]\.` matches layers 0-5 using regex syntax
+
+**Pattern Examples:**
+| Pattern | Matches |
+|---------|---------|
+| `text_model` | All text encoder layers |
+| `\.norm` | All normalization layers |
+| `attn\.(q\|k\|v)` | Query, key, value attention weights |
+| `block\.[0-9]\.` | Blocks 0-9 |

docs/merger_3_model_modes.md

@@ -0,0 +1,83 @@
+# 3-Model Merger Documentation
+
+---
+
+## Add-Difference
+> `A + (B - C) * α`. This classic technique applies the "difference" between models B and C to model A. For example, it can be used to add a style (B) to a base model (A) while subtracting the style's original base (C).
+
+**Models Used:** A, B, C
+**Parameters:**
+- **Alpha:** A multiplier for the `(B - C)` difference vector before it's added to A.
+- **Beta:** A smoothing toggle. `0` for Off, `1` for On. When On, applies a median and gaussian filter to the difference, which can result in a smoother merge.
+
+---
+
+## Train-Difference
+> A variation of Add-Difference from the SuperMerger extension that uses a more complex scaling factor based on the relative distances between the tensors of A, B, and C.
+
+**Models Used:** A, B, C
+**Parameters:**
+- **Alpha:** A multiplier for the final calculated "training" difference.
+
+---
+
+## Extract-Features
+> A powerful mode that identifies features present in both `(B - A)` and `(C - A)` and adds them to A. Allows for fine-grained control over combining aspects based on their similarity.
+
+**Models Used:** A, B, C
+**Parameters:**
+- **Alpha:** Weights the merge between Model B (`0.0`) and Model C (`1.0`).
+- **Beta:** Controls the focus on similarity (`0.0`) versus dissimilarity (`1.0`).
+- **Gamma:** A bias exponent for similarity. Higher values increase the bias.
+- **Delta:** A final multiplier for the extracted features before they are added to Model A.
+
+---
+
+## Add-Dissimilarities
+> Identifies features that are dissimilar between Model B and Model C and adds them to Model A. Useful for combining unique aspects of two different models.
+
+**Models Used:** A, B, C
+**Parameters:**
+- **Alpha:** Weights the comparison between Model B (`0.0`) and Model C (`1.0`).
+- **Beta:** A multiplier for the final extracted dissimilar features.
+- **Gamma:** A bias exponent for the similarity calculation.
+
+---
+
+## Layer Mismatch Handling
+
+When merging models with different layer structures, the `mismatch_mode` parameter controls behavior:
+
+| Mode | Behavior |
+|------|----------|
+| `skip` | Missing layers in B/C use A's values **(default)** |
+| `zeros` | Missing layers in B/C are treated as zeros |
+| `error` | Fail if any layer is missing |
+
+**Special case for B - C subtraction (Add-Difference, Train-Difference):**
+- If **both** B and C have the layer: Normal subtraction is performed
+- If **either** B or C is missing the layer: The difference is treated as `None` (no contribution from this layer), and Model A's value is used
+
+**Common scenarios:**
+- Merging a fine-tuned model with added/removed layers
+- Extracting style differences between models with slightly different architectures
+- Combining partial LoRAs or adapters
+
+**Note:** Extra layers in Models B/C that don't exist in Model A are currently ignored. The output will always have the same layer structure as Model A.
+
+---
+
+## Layer Filtering (Regex Patterns)
+
+Use regex patterns to control which layers are merged:
+
+### `exclude_patterns`
+Layers matching any pattern will **keep Model A's values only** (no merge).
+
+### `discard_patterns`
+Layers matching any pattern will be **removed entirely** from the output.
+
+**Pattern format:**
+- Whitespace-separated regex patterns
+- Patterns use **substring matching** (not full match)
+- Example: `text_model lora` matches any key containing "text_model" OR "lora"

nodes/__init__.py

@@ -1,10 +0,0 @@
-from .metakeys import *
-from .modelunloader import *
-
-__all__ = (
-    "UNetMetaKeys",
-    "CLIPMetaKeys",
-    "LoRAMetaKeys",
-    "CheckpointMetaKeys",
-    "OffloadModels",
-)

nodes/lora_extract.py

@@ -0,0 +1,350 @@
+"""
+Memory-efficient LoRA extraction node.
+
+This node loads two models from files and extracts a LoRA by computing the difference,
+without requiring the full models to be loaded into GPU memory via ComfyUI's model management.
+"""
+import os
+import torch
+import folder_paths
+import comfy.utils
+from comfy_api.latest import io
+from safetensors.torch import save_file
+from .merger_utils import MemoryEfficientSafeOpen
+
+
+CLAMP_QUANTILE = 0.99
+
+
+def extract_lora(diff, rank):
+    """Extract LoRA weights from a weight difference tensor using SVD."""
+    conv2d = (len(diff.shape) == 4)
+    kernel_size = None if not conv2d else diff.size()[2:4]
+    conv2d_3x3 = conv2d and kernel_size != (1, 1)
+    out_dim, in_dim = diff.size()[0:2]
+    rank = min(rank, in_dim, out_dim)
+
+    if conv2d:
+        if conv2d_3x3:
+            diff = diff.flatten(start_dim=1)
+        else:
+            diff = diff.squeeze()
+
+    U, S, Vh = torch.linalg.svd(diff.float())
+    U = U[:, :rank]
+    S = S[:rank]
+    U = U @ torch.diag(S)
+    Vh = Vh[:rank, :]
+
+    dist = torch.cat([U.flatten(), Vh.flatten()])
+    hi_val = torch.quantile(dist, CLAMP_QUANTILE)
+    low_val = -hi_val
+
+    U = U.clamp(low_val, hi_val)
+    Vh = Vh.clamp(low_val, hi_val)
+    if conv2d:
+        U = U.reshape(out_dim, rank, 1, 1)
+        Vh = Vh.reshape(rank, in_dim, kernel_size[0], kernel_size[1])
+    return (U, Vh)
+
+
+def extract_lora_from_files(
+    model_a_path: str,
+    model_b_path: str,
+    rank: int,
+    lora_type: str,
+    bias_diff: bool,
+    prefix_lora: str,
+    process_device: str,
+    save_dtype: str,
+) -> dict[str, torch.Tensor]:
+    """
+    Memory-efficient LoRA extraction by streaming tensors from safetensors files.
+    
+    Args:
+        model_a_path: Path to the finetuned/modified model
+        model_b_path: Path to the base/original model  
+        rank: LoRA rank for SVD decomposition
+        lora_type: "standard" for LoRA, "full_diff" for full difference
+        bias_diff: Whether to include bias differences
+        prefix_lora: Prefix for LoRA keys (e.g., "diffusion_model.")
+        process_device: Device for processing ("cuda" or "cpu")
+        save_dtype: Output dtype ("fp16", "bf16", "fp32")
+    
+    Returns:
+        Dictionary of LoRA tensors
+    """
+    output_sd = {}
+    
+    # Get dtype for saving
+    save_torch_dtype = {
+        "fp32": torch.float32, 
+        "fp16": torch.float16, 
+        "bf16": torch.bfloat16
+    }.get(save_dtype, torch.float16)
+    
+    # Open both files with memory-efficient handler
+    handler_a = MemoryEfficientSafeOpen(model_a_path)
+    handler_b = MemoryEfficientSafeOpen(model_b_path)
+    
+    try:
+        keys_a = set(handler_a.keys())
+        keys_b = set(handler_b.keys())
+        
+        # Only process keys that exist in both models
+        common_keys = keys_a & keys_b
+        weight_keys = [k for k in common_keys if k.endswith(".weight")]
+        bias_keys = [k for k in common_keys if k.endswith(".bias")] if bias_diff else []
+        
+        # Log key differences
+        missing_in_b = keys_a - keys_b
+        extra_in_b = keys_b - keys_a
+        if missing_in_b:
+            print(f"[LoRA Extract] {len(missing_in_b)} keys in model A not in model B (skipped)")
+        if extra_in_b:
+            print(f"[LoRA Extract] {len(extra_in_b)} keys in model B not in model A (ignored)")
+        
+        pbar = comfy.utils.ProgressBar(len(weight_keys) + len(bias_keys))
+        
+        for key in weight_keys:
+            # Load tensors one at a time
+            tensor_a = handler_a.get_tensor(key).to(device=process_device, dtype=torch.float32)
+            tensor_b = handler_b.get_tensor(key).to(device=process_device, dtype=torch.float32)
+            
+            # Compute difference: finetuned - base
+            weight_diff = tensor_a - tensor_b
+            
+            # Clean up immediately
+            del tensor_a, tensor_b
+            
+            # Extract LoRA key name
+            lora_key = key[:-7]  # Remove ".weight"
+            
+            if lora_type == "standard":
+                if weight_diff.ndim < 2:
+                    # 1D weights (e.g., LayerNorm) - store as diff if bias_diff enabled
+                    if bias_diff:
+                        output_sd[f"{prefix_lora}{lora_key}.diff"] = weight_diff.to(save_torch_dtype).cpu()
+                else:
+                    try:
+                        lora_up, lora_down = extract_lora(weight_diff, rank)
+                        output_sd[f"{prefix_lora}{lora_key}.lora_up.weight"] = lora_up.to(save_torch_dtype).cpu()
+                        output_sd[f"{prefix_lora}{lora_key}.lora_down.weight"] = lora_down.to(save_torch_dtype).cpu()
+                    except Exception as e:
+                        print(f"[LoRA Extract] Could not extract LoRA for {key}: {e}")
+            
+            elif lora_type == "full_diff":
+                output_sd[f"{prefix_lora}{lora_key}.diff"] = weight_diff.to(save_torch_dtype).cpu()
+            
+            del weight_diff
+            pbar.update(1)
+        
+        # Process bias keys
+        for key in bias_keys:
+            tensor_a = handler_a.get_tensor(key).to(device=process_device, dtype=torch.float32)
+            tensor_b = handler_b.get_tensor(key).to(device=process_device, dtype=torch.float32)
+            
+            bias_diff_tensor = tensor_a - tensor_b
+            del tensor_a, tensor_b
+            
+            lora_key = key[:-5]  # Remove ".bias"
+            output_sd[f"{prefix_lora}{lora_key}.diff_b"] = bias_diff_tensor.to(save_torch_dtype).cpu()
+            
+            del bias_diff_tensor
+            pbar.update(1)
+    
+    finally:
+        handler_a.__exit__(None, None, None)
+        handler_b.__exit__(None, None, None)
+    
+    return output_sd
+
+
+class LoRASaveFromFile(io.ComfyNode):
+    """
+    Memory-efficient LoRA extraction from safetensors files.
+    
+    Unlike the standard LoraSave node which requires pre-loaded and subtracted models,
+    this node loads models directly from files one tensor at a time, making it much
+    more memory efficient for extracting LoRAs from large models.
+    """
+    
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRASaveFromFile",
+            display_name="Extract LoRA from Files (Memory Efficient)",
+            category="ModelUtils/LoRA",
+            description="Extracts a LoRA by computing the difference between two model files. "
+                        "Much more memory efficient than using ModelSubtract + LoraSave.",
+            inputs=[
+                io.Combo.Input(
+                    "model_a", 
+                    options=folder_paths.get_filename_list("diffusion_models"),
+                    tooltip="The finetuned/modified model (A - B = LoRA)"
+                ),
+                io.Combo.Input(
+                    "model_b", 
+                    options=folder_paths.get_filename_list("diffusion_models"),
+                    tooltip="The base/original model (A - B = LoRA)"
+                ),
+                io.Int.Input("rank", default=64, min=1, max=4096, step=1,
+                            tooltip="LoRA rank - higher preserves more detail but increases size"),
+                io.Combo.Input("lora_type", options=["standard", "full_diff"], default="standard",
+                              tooltip="standard: SVD decomposition to lora_up/lora_down, full_diff: raw weight differences"),
+                io.Boolean.Input("bias_diff", default=True,
+                                tooltip="Include bias differences in the LoRA"),
+                io.String.Input("output_filename", default="extracted_lora",
+                               tooltip="Output filename (without extension)"),
+                io.Combo.Input("save_dtype", options=["fp16", "bf16", "fp32"], default="fp16"),
+                io.Combo.Input("process_device", options=["cuda", "cpu"], default="cuda",
+                              tooltip="Device for SVD computation"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+            is_output_node=True,
+        )
+
+    @classmethod
+    def execute(
+        cls,
+        model_a: str,
+        model_b: str,
+        rank: int,
+        lora_type: str,
+        bias_diff: bool,
+        output_filename: str,
+        save_dtype: str,
+        process_device: str,
+    ) -> io.NodeOutput:
+        # Get full paths
+        model_a_path = folder_paths.get_full_path_or_raise("diffusion_models", model_a)
+        model_b_path = folder_paths.get_full_path_or_raise("diffusion_models", model_b)
+        
+        # Extract LoRA
+        output_sd = extract_lora_from_files(
+            model_a_path=model_a_path,
+            model_b_path=model_b_path,
+            rank=rank,
+            lora_type=lora_type,
+            bias_diff=bias_diff,
+            prefix_lora="diffusion_model.",
+            process_device=process_device,
+            save_dtype=save_dtype,
+        )
+        
+        if not output_sd:
+            raise ValueError("No LoRA weights extracted - models may be identical or incompatible")
+        
+        # Save to loras folder
+        output_dir = folder_paths.get_folder_paths("loras")[0]
+        os.makedirs(output_dir, exist_ok=True)
+        output_path = os.path.join(output_dir, f"{output_filename.strip()}.safetensors")
+        
+        save_file(output_sd, output_path)
+        print(f"[LoRA Extract] Saved {len(output_sd)} tensors to {output_path}")
+        
+        return io.NodeOutput(output_path)
+
+
+class LoRACheckpointSaveFromFile(io.ComfyNode):
+    """
+    Memory-efficient LoRA extraction from checkpoint files (which may contain both model and text encoder).
+    """
+    
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRACheckpointSaveFromFile",
+            display_name="Extract LoRA from Checkpoints (Memory Efficient)",
+            category="ModelUtils/LoRA",
+            description="Extracts a LoRA by computing the difference between two checkpoint files. "
+                        "Handles both diffusion model and text encoder weights.",
+            inputs=[
+                io.Combo.Input(
+                    "checkpoint_a", 
+                    options=folder_paths.get_filename_list("checkpoints"),
+                    tooltip="The finetuned/modified checkpoint (A - B = LoRA)"
+                ),
+                io.Combo.Input(
+                    "checkpoint_b", 
+                    options=folder_paths.get_filename_list("checkpoints"),
+                    tooltip="The base/original checkpoint (A - B = LoRA)"
+                ),
+                io.Int.Input("rank", default=64, min=1, max=4096, step=1,
+                            tooltip="LoRA rank - higher preserves more detail but increases size"),
+                io.Combo.Input("lora_type", options=["standard", "full_diff"], default="standard",
+                              tooltip="standard: SVD decomposition, full_diff: raw weight differences"),
+                io.Boolean.Input("bias_diff", default=True,
+                                tooltip="Include bias differences in the LoRA"),
+                io.String.Input("output_filename", default="extracted_lora",
+                               tooltip="Output filename (without extension)"),
+                io.Combo.Input("save_dtype", options=["fp16", "bf16", "fp32"], default="fp16"),
+                io.Combo.Input("process_device", options=["cuda", "cpu"], default="cuda",
+                              tooltip="Device for SVD computation"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+            is_output_node=True,
+        )
+
+    @classmethod
+    def execute(
+        cls,
+        checkpoint_a: str,
+        checkpoint_b: str,
+        rank: int,
+        lora_type: str,
+        bias_diff: bool,
+        output_filename: str,
+        save_dtype: str,
+        process_device: str,
+    ) -> io.NodeOutput:
+        # Get full paths
+        checkpoint_a_path = folder_paths.get_full_path_or_raise("checkpoints", checkpoint_a)
+        checkpoint_b_path = folder_paths.get_full_path_or_raise("checkpoints", checkpoint_b)
+        
+        # Open handlers to detect prefixes
+        handler_a = MemoryEfficientSafeOpen(checkpoint_a_path)
+        keys = handler_a.keys()
+        handler_a.__exit__(None, None, None)
+        
+        # Auto-detect key prefixes
+        has_model = any(k.startswith("model.") for k in keys)
+        has_diffusion = any(k.startswith("diffusion_model.") for k in keys)
+        has_text_encoders = any(k.startswith("text_encoders.") or k.startswith("cond_stage_model.") for k in keys)
+        
+        output_sd = {}
+        
+        # Extract diffusion model LoRA
+        if has_model or has_diffusion:
+            prefix = "model.diffusion_model." if has_model else "diffusion_model."
+            model_lora = extract_lora_from_files(
+                model_a_path=checkpoint_a_path,
+                model_b_path=checkpoint_b_path,
+                rank=rank,
+                lora_type=lora_type,
+                bias_diff=bias_diff,
+                prefix_lora="diffusion_model.",
+                process_device=process_device,
+                save_dtype=save_dtype,
+            )
+            output_sd.update(model_lora)
+        
+        # Note: Text encoder extraction would need additional prefix handling
+        # For full checkpoint support, this would need to be expanded
+        
+        if not output_sd:
+            raise ValueError("No LoRA weights extracted - checkpoints may be identical or incompatible")
+        
+        # Save to loras folder
+        output_dir = folder_paths.get_folder_paths("loras")[0]
+        os.makedirs(output_dir, exist_ok=True)
+        output_path = os.path.join(output_dir, f"{output_filename.strip()}.safetensors")
+        
+        save_file(output_sd, output_path)
+        print(f"[LoRA Extract] Saved {len(output_sd)} tensors to {output_path}")
+        
+        return io.NodeOutput(output_path)

nodes/merger.py

@@ -0,0 +1,715 @@
+import os
+import re
+import torch
+import folder_paths
+import comfy.utils
+from comfy_api.latest import io
+from safetensors.torch import save_file
+from .merger_ops import TWO_MODEL_MODES, THREE_MODEL_MODES, MissingTensorBehavior, MissingTensorError
+from .merger_utils import MemoryEfficientSafeOpen
+
+
+def load_documentation_from_file(filename):
+    """Loads documentation from a markdown file in the ../docs/ directory."""
+    current_dir = os.path.dirname(os.path.realpath(__file__))
+    docs_path = os.path.join(current_dir, '..', 'docs', filename)
+    try:
+        with open(docs_path, 'r', encoding='utf-8') as f:
+            return f.read()
+    except FileNotFoundError:
+        return f"# Documentation File Not Found\n\nPlease ensure `{filename}` exists in the `docs` directory of the custom node."
+
+
+def _compile_patterns(pattern_string):
+    """Compiles whitespace-separated regex patterns into a list of compiled patterns.
+    
+    Returns empty list if pattern_string is empty or whitespace-only.
+    Raises ValueError if any pattern is invalid regex.
+    """
+    if not pattern_string or not pattern_string.strip():
+        return []
+    
+    patterns = []
+    for pattern in pattern_string.split():
+        try:
+            patterns.append(re.compile(pattern))
+        except re.error as e:
+            raise ValueError(f"Invalid regex pattern '{pattern}': {e}")
+    return patterns
+
+
+def _matches_any_pattern(key, patterns):
+    """Returns True if key matches any of the compiled regex patterns (substring search)."""
+    for pattern in patterns:
+        if pattern.search(key):
+            return True
+    return False
+
+
+class MergerLogic:
+    """Shared merge execution logic for all merger nodes."""
+    
+    @staticmethod
+    def execute_merge(model_names, calc_mode, all_modes, recipe_params, model_type):
+        calc_mode_class = next((m for m in all_modes if m.name == calc_mode), None)
+        if not calc_mode_class:
+            raise ValueError(f"Calc mode '{calc_mode}' not found.")
+        primary_model_name = model_names.get('model_a')
+        if not primary_model_name or primary_model_name == "None":
+            raise ValueError("Model A is required to run the merge.")
+        
+        handlers = {}
+        for name in model_names.values():
+            if name and name != "None":
+                path = folder_paths.get_full_path(model_type, name)
+                if not path:
+                    raise FileNotFoundError(f"Model '{name}' not found.")
+                handlers[name] = MemoryEfficientSafeOpen(path)
+        
+        primary_handler = handlers[primary_model_name]
+        all_keys = primary_handler.keys()
+        metadata = primary_handler.header.get("__metadata__", {})
+        
+        # Convert mismatch_mode string to enum
+        mismatch_mode_str = recipe_params.get('mismatch_mode', 'skip')
+        mismatch_mode = MissingTensorBehavior(mismatch_mode_str)
+        recipe_params['mismatch_mode'] = mismatch_mode
+        
+        # Compile filter patterns
+        exclude_patterns = _compile_patterns(recipe_params.get('exclude_patterns', ''))
+        discard_patterns = _compile_patterns(recipe_params.get('discard_patterns', ''))
+        
+        # Pre-compute key differences for logging
+        primary_keys = set(all_keys)
+        for name, handler in handlers.items():
+            if name != primary_model_name:
+                secondary_keys = set(handler.keys())
+                missing = primary_keys - secondary_keys
+                extra = secondary_keys - primary_keys
+                if missing:
+                    print(f"[Merger] {name} is missing {len(missing)} keys present in Model A")
+                if extra:
+                    print(f"[Merger] {name} has {len(extra)} extra keys not in Model A (ignored)")
+        
+        merged_state_dict = {}
+        pbar = comfy.utils.ProgressBar(len(all_keys))
+        save_dtype = recipe_params.pop('save_dtype')
+        save_torch_dtype = {"fp32": torch.float32, "fp16": torch.float16, "bf16": torch.bfloat16}.get(save_dtype)
+        recipe_params.update({"handlers": handlers})
+        process_device = recipe_params.get('device', 'cpu')
+        process_dtype = recipe_params.get('dtype', torch.float32)
+        
+        skipped_keys = 0
+        excluded_keys = 0
+        discarded_keys = 0
+        error_keys = []
+        
+        for key in all_keys:
+            # Check discard patterns first - skip entirely
+            if _matches_any_pattern(key, discard_patterns):
+                discarded_keys += 1
+                pbar.update(1)
+                continue
+            
+            # Pre-load Model A's tensor
+            tensor_a = primary_handler.get_tensor(key).to(device=process_device, dtype=process_dtype)
+            
+            # Check exclude patterns - use Model A only, no merge
+            if _matches_any_pattern(key, exclude_patterns):
+                merged_state_dict[key] = tensor_a.to(save_torch_dtype).cpu().clone()
+                excluded_keys += 1
+                pbar.update(1)
+                continue
+            
+            # Pass tensor_a metadata to recipes for zeros mode and fallback
+            recipe_params['_tensor_a'] = tensor_a
+            recipe_params['_tensor_a_shape'] = tensor_a.shape
+            recipe_params['_tensor_a_dtype'] = tensor_a.dtype
+            
+            try:
+                recipe = calc_mode_class.create_recipe(key=key, **recipe_params)
+                result = recipe.merge()
+            except MissingTensorError as e:
+                if mismatch_mode == MissingTensorBehavior.ERROR:
+                    raise ValueError(f"Layer mismatch error (mismatch_mode='error'): {e}")
+                result = None
+                error_keys.append(key)
+            
+            # Handle None result (mismatch occurred with skip mode)
+            if result is None:
+                result = tensor_a
+                skipped_keys += 1
+            
+            if isinstance(result, dict):
+                for r_key, r_tensor in result.items():
+                    merged_state_dict[r_key] = r_tensor.to(save_torch_dtype).cpu().clone()
+            else:
+                merged_state_dict[key] = result.to(save_torch_dtype).cpu().clone()
+            
+            # Clean up tensor_a reference to allow GC
+            del recipe_params['_tensor_a']
+            pbar.update(1)
+        
+        # Log summary
+        if excluded_keys > 0:
+            print(f"[Merger] Excluded {excluded_keys} keys from merge (kept Model A only)")
+        if discarded_keys > 0:
+            print(f"[Merger] Discarded {discarded_keys} keys from output")
+        if skipped_keys > 0:
+            print(f"[Merger] Used Model A's values for {skipped_keys} keys due to mismatches")
+        if error_keys:
+            print(f"[Merger] Unexpected errors on {len(error_keys)} keys: {error_keys[:5]}{'...' if len(error_keys) > 5 else ''}")
+        
+        for handler in handlers.values():
+            handler.__exit__(None, None, None)
+        output_folder = "loras" if calc_mode == "SVD LoRA Extraction" else model_type
+        output_dir = folder_paths.get_folder_paths(output_folder)[0]
+        os.makedirs(output_dir, exist_ok=True)
+        output_filename = recipe_params.get("output_filename")
+        output_path = os.path.join(output_dir, f"{output_filename}.safetensors")
+        save_file(merged_state_dict, output_path, metadata=metadata)
+        return f"{output_filename}.safetensors"
+
+
+# --- Two-Model Merger Nodes ---
+
+class CheckpointTwoMerger(io.ComfyNode):
+    MODEL_TYPE = "checkpoints"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CheckpointTwoMerger",
+            display_name="Merge Checkpoints (2 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("checkpoints")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("checkpoints")),
+                io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_2_checkpoint"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_2_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class ModelTwoMerger(io.ComfyNode):
+    MODEL_TYPE = "diffusion_models"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="ModelTwoMerger",
+            display_name="Merge Models (2 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
+                io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_2_model"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_2_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class TextEncoderTwoMerger(io.ComfyNode):
+    MODEL_TYPE = "text_encoders"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="TextEncoderTwoMerger",
+            display_name="Merge Text Encoders (2 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("text_encoders")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("text_encoders")),
+                io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_2_textencoder"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_2_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class LoRATwoMerger(io.ComfyNode):
+    MODEL_TYPE = "loras"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRATwoMerger",
+            display_name="Merge LoRAs (2 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("loras")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("loras")),
+                io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_2_lora"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_2_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class EmbeddingTwoMerger(io.ComfyNode):
+    MODEL_TYPE = "embeddings"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="EmbeddingTwoMerger",
+            display_name="Merge Embeddings (2 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("embeddings")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("embeddings")),
+                io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_2_embedding"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_2_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+# --- Three-Model Merger Nodes ---
+
+class CheckpointThreeMerger(io.ComfyNode):
+    MODEL_TYPE = "checkpoints"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CheckpointThreeMerger",
+            display_name="Merge Checkpoints (3 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("checkpoints")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("checkpoints")),
+                io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("checkpoints")),
+                io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_3_checkpoint"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_3_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class ModelThreeMerger(io.ComfyNode):
+    MODEL_TYPE = "diffusion_models"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="ModelThreeMerger",
+            display_name="Merge Models (3 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
+                io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
+                io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_3_model"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_3_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class TextEncoderThreeMerger(io.ComfyNode):
+    MODEL_TYPE = "text_encoders"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="TextEncoderThreeMerger",
+            display_name="Merge Text Encoders (3 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("text_encoders")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("text_encoders")),
+                io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("text_encoders")),
+                io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_3_textencoder"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_3_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class LoRAThreeMerger(io.ComfyNode):
+    MODEL_TYPE = "loras"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRAThreeMerger",
+            display_name="Merge LoRAs (3 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("loras")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("loras")),
+                io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("loras")),
+                io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_3_lora"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_3_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)
+
+
+class EmbeddingThreeMerger(io.ComfyNode):
+    MODEL_TYPE = "embeddings"
+
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="EmbeddingThreeMerger",
+            display_name="Merge Embeddings (3 Models)",
+            category="ModelUtils/Merging",
+            inputs=[
+                io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
+                io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("embeddings")),
+                io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("embeddings")),
+                io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("embeddings")),
+                io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
+                io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
+                io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
+                io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
+                io.String.Input("output_filename", default="merged_3_embedding"),
+                io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
+                io.Combo.Input("process_device", options=["cuda", "cpu"]),
+                io.String.Input("exclude_patterns", default="", multiline=True),
+                io.String.Input("discard_patterns", default="", multiline=True),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_filename"),
+                io.String.Output(display_name="documentation"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
+                calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
+                gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
+                process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
+        doc = load_documentation_from_file('merger_3_model_modes.md')
+        if execution_mode == "DOCUMENTATION ONLY":
+            return io.NodeOutput("Documentation mode active. No merge performed.", doc)
+
+        recipe_params = {
+            "model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
+            "mismatch_mode": mismatch_mode,
+            "alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
+            "output_filename": output_filename, "save_dtype": save_dtype,
+            "device": process_device, "dtype": torch.float32,
+            "exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
+        }
+        model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
+        filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
+        return io.NodeOutput(filename, doc)

nodes/merger_ops.py

@@ -0,0 +1,408 @@
+import torch
+import scipy.ndimage
+import torch.nn.functional as F
+from enum import Enum
+
+
+class MissingTensorBehavior(Enum):
+    """Controls behavior when a tensor key is missing from a model."""
+    ERROR = "error"      # Raise exception (strict mode)
+    SKIP = "skip"        # Return None, let caller handle
+    ZEROS = "zeros"      # Return zeros tensor matching shape from primary model
+
+
+class MissingTensorError(Exception):
+    """Raised when a tensor is missing and behavior is ERROR."""
+    pass
+
+# ################# #
+# MERGE OPERATORS
+# ################# #
+class Operation:
+    def __init__(self, key, *sources):
+        self.key = key
+        self.sources = tuple(sources)
+        self.merge_func = self.recurse
+        self.alpha = None
+        self.beta = None
+        self.gamma = None
+        self.delta = None
+        self.seed = None
+
+    def __eq__(self, other):
+        return (type(self), self.key, self.alpha, self.beta, self.gamma, self.delta, self.seed, self.sources) == \
+               (type(other), other.key, other.alpha, other.beta, other.gamma, other.delta, other.seed, other.sources)
+
+    def __hash__(self):
+        return hash((type(self), self.key, self.alpha, self.beta, self.gamma, self.delta, self.seed, self.sources))
+
+    def oper(self, *args) -> torch.Tensor:
+        raise NotImplementedError
+
+    def recurse(self, operation):
+        source_tensors = [source_oper.merge() for source_oper in operation.sources]
+        return operation.oper(*source_tensors)
+
+    def merge(self):
+        return self.merge_func(self)
+
+class LoadTensor(Operation):
+    def __init__(self, key, model_name, handlers, device, dtype,
+                 on_missing=MissingTensorBehavior.ERROR, fallback_shape=None, fallback_dtype=None):
+        super().__init__(key)
+        self.model_name = model_name
+        self.handlers = handlers
+        self.device = device
+        self.dtype = dtype
+        self.on_missing = on_missing
+        self.fallback_shape = fallback_shape
+        self.fallback_dtype = fallback_dtype
+
+    def merge(self) -> torch.Tensor:
+        if self.model_name not in self.handlers:
+            raise ValueError(f"Model '{self.model_name}' is required for this mode but was not provided.")
+        
+        handler = self.handlers[self.model_name]
+        if self.key not in handler.keys():
+            if self.on_missing == MissingTensorBehavior.ERROR:
+                raise MissingTensorError(
+                    f"Key '{self.key}' not found in model '{self.model_name}'"
+                )
+            elif self.on_missing == MissingTensorBehavior.SKIP:
+                return None
+            elif self.on_missing == MissingTensorBehavior.ZEROS:
+                if self.fallback_shape is None:
+                    raise ValueError(
+                        f"Cannot create zeros for '{self.key}' without fallback_shape"
+                    )
+                dtype = self.fallback_dtype if self.fallback_dtype else self.dtype
+                return torch.zeros(self.fallback_shape, device=self.device, dtype=dtype)
+        
+        return handler.get_tensor(self.key).to(device=self.device, dtype=self.dtype)
+
+class Multiply(Operation):
+    def __init__(self, key, alpha, *sources):
+        super().__init__(key, *sources)
+        self.alpha = alpha
+    def oper(self, a) -> torch.Tensor:
+        if a is None:
+            return None
+        return a * self.alpha
+
+
+class Add(Operation):
+    def oper(self, a, b) -> torch.Tensor:
+        if a is None:
+            return b
+        if b is None:
+            return a
+        return a + b
+
+
+class Sub(Operation):
+    def oper(self, a, b) -> torch.Tensor:
+        if a is None or b is None:
+            return None  # Cannot compute difference with missing operand
+        return a - b
+
+class SVD(Operation):
+    def __init__(self, key, alpha, beta, gamma, *sources):
+        super().__init__(key, *sources)
+        self.alpha, self.beta, self.gamma = int(alpha), int(beta), gamma
+    def oper(self, a, b) -> torch.Tensor:
+        if a is None or b is None:
+            return None
+        diff, weights, conv2d = a - b, {}, len(a.size()) == 4
+        kernel_size = None if not conv2d else a.size()[2:4]
+        conv2d_3x3 = conv2d and kernel_size != (1, 1)
+        rank = self.alpha if not conv2d_3x3 or self.beta is None else self.beta
+        rank = min(rank, a.size()[1], a.size()[0])
+        matrix = diff.flatten(start_dim=1) if conv2d_3x3 else diff.squeeze() if conv2d else diff
+        U, S, Vh = torch.linalg.svd(matrix)
+        U, S, Vh = U[:, :rank], S[:rank], Vh[:rank, :]
+        U = U @ torch.diag(S)
+        dist = torch.cat([U.flatten(), Vh.flatten()])
+        hi_val = torch.quantile(dist, self.gamma)
+        U, Vh = U.clamp(-hi_val, hi_val), Vh.clamp(-hi_val, hi_val)
+        if conv2d:
+            Vh = Vh.reshape(rank, a.size()[1], *kernel_size) if conv2d_3x3 else Vh.reshape(rank, a.size()[1], 1, 1)
+            U = U.reshape(a.size()[0], rank, 1, 1)
+        weights_sd = {f"{self.key}.lora_up.weight": U, f"{self.key}.lora_down.weight": Vh, f"{self.key}.alpha": torch.tensor(float(rank))}
+        return weights_sd
+
+class Smooth(Operation):
+    def oper(self, a) -> torch.Tensor:
+        if a is None:
+            return None
+        filtered = scipy.ndimage.median_filter(a.detach().cpu().to(torch.float32).numpy(), size=3)
+        filtered = scipy.ndimage.gaussian_filter(filtered, sigma=1)
+        return torch.tensor(filtered, dtype=a.dtype, device=a.device)
+class TrainDiff(Operation):
+    def oper(self, a, b, c) -> torch.Tensor:
+        if a is None or b is None or c is None:
+            return None
+        if torch.allclose(b.float(), c.float()): return torch.zeros_like(a)
+        diff_bc = b.float() - c.float()
+        dist_bc, dist_ab = torch.abs(diff_bc), torch.abs(b.float() - a.float())
+        scale = torch.where(dist_bc + dist_ab != 0, dist_ab / (dist_bc + dist_ab), 0.0)
+        return (torch.sign(diff_bc) * torch.abs(scale) * torch.abs(diff_bc)).to(a.dtype) * 1.8
+class ExtractOp(Operation):
+    def __init__(self, key, alpha, beta, gamma, *args):
+        super().__init__(key, *args)
+        self.alpha, self.beta, self.gamma = alpha, beta, gamma
+    def oper(self, base: torch.Tensor, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
+        if a is None or b is None:
+            return None
+        base_f = base.float() if base is not None else 0
+        a_f, b_f = a.float() - base_f, b.float() - base_f
+        c = torch.cosine_similarity(a_f.flatten(), b_f.flatten(), 0).clamp(-1, 1)
+        d = ((c + 1) / 2)**self.gamma
+        return (torch.lerp(a_f, b_f, self.alpha) * torch.lerp(d, 1 - d, self.beta)).to(a.dtype)
+
+
+class Similarities(ExtractOp):
+    def oper(self, a: torch.Tensor, b: torch.Tensor) -> torch.Tensor:
+        if a is None or b is None:
+            return None
+        return super().oper(None, a, b)
+class PowerUpOp(Operation):
+    def __init__(self, key, alpha, seed, *sources):
+        super().__init__(key, *sources)
+        self.alpha, self.seed = alpha, seed
+    def oper(self, a, b):
+        if a is None or b is None:
+            return None
+        delta = b - a
+        rng = torch.Generator(device=delta.device).manual_seed(self.seed)
+        m = torch.bernoulli(torch.full_like(delta, 1 - self.alpha), generator=rng)
+        return (m * delta) / (1 - self.alpha) if (1 - self.alpha) > 0 else torch.zeros_like(delta)
+
+
+class InterpolateDifference(Operation):
+    def __init__(self, key, alpha, beta, gamma, seed, *sources):
+        super().__init__(key, *sources)
+        self.alpha, self.beta, self.gamma, self.seed = alpha, beta, gamma, seed
+    def oper(self, a, b):
+        if a is None or b is None:
+            return None
+        alpha, delta = max(self.alpha, 0.001), torch.abs(a - b)
+        max_delta = torch.max(delta)
+        if max_delta == 0: return a
+        diff = ((max_delta - delta) / max_delta if self.beta != 1 else delta / max_delta) ** (1 / alpha - 1)
+        diff = torch.nan_to_num(diff)
+        rng = torch.Generator(device=diff.device).manual_seed(self.seed)
+        mask = torch.lerp(torch.bernoulli(torch.clamp(diff, 0, 1), generator=rng), diff, self.gamma)
+        return a * (1 - mask) + b * mask
+
+# ################# #
+# CALCULATION MODES
+# ################# #
+class CalcMode:
+    name = 'calcmode'; description = 'description'; models_used = []; param_docs = {}
+    def create_recipe(self, key, **kwargs) -> Operation: raise NotImplementedError
+    
+    @staticmethod
+    def _get_secondary_loader_args(kwargs, mismatch_mode):
+        """Helper to build loader args for secondary models (B, C) with fallback support."""
+        base_args = {
+            "handlers": kwargs['handlers'],
+            "device": kwargs['device'],
+            "dtype": kwargs['dtype'],
+            "on_missing": mismatch_mode,
+        }
+        # Add fallback info for zeros mode
+        if '_tensor_a_shape' in kwargs:
+            base_args['fallback_shape'] = kwargs['_tensor_a_shape']
+            base_args['fallback_dtype'] = kwargs.get('_tensor_a_dtype', kwargs['dtype'])
+        return base_args
+
+
+class PreloadedTensor(Operation):
+    """Returns a pre-loaded tensor instead of loading from file.
+    
+    Used when Model A's tensor is pre-loaded in the merge loop.
+    """
+    def __init__(self, key, tensor):
+        super().__init__(key)
+        self.tensor = tensor
+    
+    def merge(self) -> torch.Tensor:
+        return self.tensor
+
+class WeightSum(CalcMode):
+    name = 'Weight-Sum'
+    description = 'A * (1 - α) + B * α. Simple linear interpolation.'
+    models_used = ['A', 'B']
+    param_docs = {'alpha': 'Interpolation weight. 0.0 is 100% Model A, 1.0 is 100% Model B.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        # Use pre-loaded tensor_a if available
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_b = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_b)
+        
+        if kwargs['alpha'] >= 1: return b
+        if kwargs['alpha'] <= 0: return a
+        c = Multiply(key, 1 - kwargs['alpha'], a)
+        d = Multiply(key, kwargs['alpha'], b)
+        return Add(key, c, d)
+
+
+class AddDifference(CalcMode):
+    name = 'Add-Difference'
+    description = 'A + (B - C) * α. Applies the difference between B and C to A.'
+    models_used = ['A', 'B', 'C']
+    param_docs = {'alpha': 'Multiplier for the (B - C) difference.', 'beta': 'Smoothing (0=Off, 1=On).'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_bc = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_bc)
+        c = LoadTensor(key, kwargs['model_c'], **loader_args_bc)
+        diff = Sub(key, b, c)
+        if kwargs['beta'] == 1: diff = Smooth(key, diff)
+        diffm = Multiply(key, kwargs['alpha'], diff)
+        return Add(key, a, diffm)
+
+
+class TrainDifference(CalcMode):
+    name = 'Train-Difference'
+    description = 'A + (B - C) * α, using tensor distances.'
+    models_used = ['A', 'B', 'C']
+    param_docs = {'alpha': 'Multiplier for the calculated difference.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_bc = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_bc)
+        c = LoadTensor(key, kwargs['model_c'], **loader_args_bc)
+        diff = TrainDiff(key, a, b, c)
+        diffm = Multiply(key, kwargs['alpha'], diff)
+        return Add(key, a, diffm)
+class InterpDifference(CalcMode):
+    name = 'Comparative-Interpolation'
+    description = 'Interpolates A and B based on relative tensor value differences.'
+    models_used = ['A', 'B']
+    param_docs = {'alpha': 'Curve strength.', 'beta': 'Style (0=Similarity, 1=Difference).', 'gamma': 'Mix (0=Random, 1=Linear).'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        # Skip certain keys entirely
+        if key.startswith('cond_stage_model.transformer.text_model.embeddings'):
+            return a
+        
+        loader_args_b = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_b)
+        return InterpolateDifference(key, kwargs['alpha'], kwargs['beta'], kwargs['gamma'], kwargs['seed'], a, b)
+
+
+class Extract(CalcMode):
+    name = 'Extract-Features'
+    description = 'Adds features from B and C to A, based on their similarity.'
+    models_used = ['A', 'B', 'C']
+    param_docs = {'alpha': 'Weighting B vs C.', 'beta': 'Similarity vs dissimilarity.', 'gamma': 'Similarity bias.', 'delta': 'Multiplier for final features.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_bc = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_bc)
+        c = LoadTensor(key, kwargs['model_c'], **loader_args_bc)
+        extracted = ExtractOp(key, kwargs['alpha'], kwargs['beta'], kwargs['gamma'] * 15, a, b, c)
+        multiplied = Multiply(key, kwargs['delta'], extracted)
+        return Add(key, a, multiplied)
+
+
+class AddDisimilarity(CalcMode):
+    name = 'Add-Dissimilarities'
+    description = 'Adds dissimilar features between B and C to model A.'
+    models_used = ['A', 'B', 'C']
+    param_docs = {'alpha': 'Weighting B vs C.', 'beta': 'Multiplier for final features.', 'gamma': 'Similarity bias.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_bc = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_bc)
+        c = LoadTensor(key, kwargs['model_c'], **loader_args_bc)
+        extracted = Similarities(key, kwargs['alpha'], 1, kwargs['gamma'] * 15, b, c)
+        multiplied = Multiply(key, kwargs['beta'], extracted)
+        return Add(key, a, multiplied)
+
+
+class PowerUp(CalcMode):
+    name = 'Power-Up (DARE)'
+    description = 'Adds capabilities of B to A using Drop and Rescale (DARE).'
+    models_used = ['A', 'B']
+    param_docs = {'alpha': 'Dropout rate (proportion of weights from B to *drop*).', 'beta': 'Multiplier for the final difference.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_b = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_b)
+        deltahat = PowerUpOp(key, kwargs['alpha'], kwargs['seed'], a, b)
+        res = Multiply(key, kwargs['beta'], deltahat)
+        return Add(key, a, res)
+
+
+class SVDMode(CalcMode):
+    name = 'SVD LoRA Extraction'
+    description = 'Extracts the difference between B and A into a LoRA file.'
+    models_used = ['A', 'B']
+    param_docs = {'alpha': 'Rank for standard layers.', 'beta': 'Rank for 3x3 conv layers.', 'gamma': 'Clamp quantile for weights.'}
+    def create_recipe(self, key, **kwargs):
+        mismatch_mode = kwargs.get('mismatch_mode', MissingTensorBehavior.SKIP)
+        
+        if '_tensor_a' in kwargs:
+            a = PreloadedTensor(key, kwargs['_tensor_a'])
+        else:
+            loader_args_a = {"handlers": kwargs['handlers'], "device": kwargs['device'], "dtype": kwargs['dtype']}
+            a = LoadTensor(key, kwargs['model_a'], **loader_args_a)
+        
+        loader_args_b = self._get_secondary_loader_args(kwargs, mismatch_mode)
+        b = LoadTensor(key, kwargs['model_b'], **loader_args_b)
+        return SVD(key, kwargs['alpha'], kwargs['beta'], kwargs['gamma'], a, b)
+
+
+TWO_MODEL_MODES = [WeightSum(), InterpDifference(), PowerUp(), SVDMode()]
+THREE_MODEL_MODES = [AddDifference(), TrainDifference(), Extract(), AddDisimilarity()]

nodes/merger_utils.py

@@ -0,0 +1,106 @@
+import json
+import struct
+import torch
+
+class MemoryEfficientSafeOpen:
+    def __init__(self, filename, device='cpu'):
+        self.filename = filename
+        self.device = device
+        self.header, self.header_size = self._read_header()
+        self.file = open(filename, "rb")
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.file.close()
+
+    def keys(self):
+        return [k for k in self.header.keys() if k != "__metadata__"]
+
+    def get_tensor(self, key):
+        if key not in self.header:
+            raise KeyError(f"Tensor '{key}' not found in the file")
+
+        metadata = self.header[key]
+        offset_start, offset_end = metadata["data_offsets"]
+
+        tensor_bytes = None
+        if offset_start != offset_end:
+            self.file.seek(self.header_size + 8 + offset_start)
+            tensor_bytes = self.file.read(offset_end - offset_start)
+
+        return self._deserialize_tensor(tensor_bytes, metadata)
+
+    def _read_header(self):
+        with open(self.filename, "rb") as f:
+            header_size = struct.unpack("<Q", f.read(8))[0]
+            header_json = f.read(header_size).decode("utf-8")
+        return json.loads(header_json), header_size
+
+    def _deserialize_tensor(self, tensor_bytes, metadata):
+        dtype_str = metadata["dtype"]
+        shape = metadata["shape"]
+        dtype = self._get_torch_dtype(dtype_str)
+
+        if tensor_bytes is None:
+            byte_tensor = torch.empty(0, dtype=torch.uint8)
+        else:
+            byte_tensor = torch.frombuffer(bytearray(tensor_bytes), dtype=torch.uint8)
+
+        if dtype_str in ["F8_E5M2", "F8_E4M3"]:
+            return self._convert_float8(byte_tensor, dtype_str, shape)
+
+        return byte_tensor.view(dtype).reshape(shape)
+
+    @staticmethod
+    def _get_torch_dtype(dtype_str):
+        dtype_map = {
+            "F64": torch.float64, "F32": torch.float32, "F16": torch.float16, "BF16": torch.bfloat16,
+            "I64": torch.int64, "I32": torch.int32, "I16": torch.int16, "I8": torch.int8,
+            "U8": torch.uint8, "BOOL": torch.bool,
+        }
+        if hasattr(torch, "float8_e5m2"):
+            dtype_map["F8_E5M2"] = torch.float8_e5m2
+        if hasattr(torch, "float8_e4m3fn"):
+            dtype_map["F8_E4M3"] = torch.float8_e4m3fn
+
+        dtype = dtype_map.get(dtype_str)
+        if dtype is None:
+            raise ValueError(f"Unsupported dtype: {dtype_str}")
+        return dtype
+
+    @staticmethod
+    def _convert_float8(byte_tensor, dtype_str, shape):
+        if dtype_str == "F8_E5M2" and hasattr(torch, "float8_e5m2"):
+            return byte_tensor.view(torch.float8_e5m2).reshape(shape)
+        elif dtype_str == "F8_E4M3" and hasattr(torch, "float8_e4m3fn"):
+            return byte_tensor.view(torch.float8_e4m3fn).reshape(shape)
+        else:
+            raise ValueError(f"Unsupported float8 type: {dtype_str}. Your PyTorch version may be too old.")
+
+
+def get_union_keys(*handlers):
+    """Returns the union of all tensor keys across multiple model handlers.
+    
+    Useful when you want to iterate over all unique keys present in any model.
+    """
+    all_keys = set()
+    for handler in handlers:
+        if handler is not None:
+            all_keys.update(handler.keys())
+    return sorted(all_keys)
+
+
+def get_intersection_keys(*handlers):
+    """Returns only keys present in ALL model handlers.
+    
+    Useful when you want to iterate only over keys that exist in every model.
+    """
+    handlers = [h for h in handlers if h is not None]
+    if not handlers:
+        return []
+    all_keys = set(handlers[0].keys())
+    for handler in handlers[1:]:
+        all_keys &= set(handler.keys())
+    return sorted(all_keys)

nodes/metakeys.py

@@ -1,118 +1,159 @@
-import torch
-from safetensors import safe_open
-from safetensors.torch import save_file
 import os
 import folder_paths
-from comfy.model_management import get_torch_device
+from comfy_api.latest import io
+from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
 
 
-def convert_pt_to_safetensors(pt_path, safe_path):
-    try:
-        model = torch.load(pt_path, map_location="cpu")
-        if "state_dict" in model:
-            state_dict = model["state_dict"]
-        elif "model" in model:
-            state_dict = model["model"]
-        else:
-            state_dict = model
-        state_dict = {k.replace("state_dict.", ""): v for k, v in state_dict.items()}
-        metadata = {"format": "pt"}
-        save_file(state_dict, safe_path, metadata)
-        return True, ""
-    except Exception as e:
-        return False, str(e)
+def _get_metakeys(model_name: str, model_type: str) -> tuple[str, str]:
+    """Shared logic for all MetaKeys nodes."""
+    model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
+    model_weights = {}
+    metadata = {}
+    layer_shapes = ""
 
-def load_metadata_from_safetensors(safetensors_file: str) -> dict:
-    try:
-        with safe_open(safetensors_file, framework="pt", device="cpu") as f:
-            model_weights = {}
-            for key in f.keys():
-                model_weights[key] = f.get_tensor(key)
-            metadata = f.metadata()
-        if metadata is None:
-            metadata = {}
-        return model_weights, metadata
-    except Exception as e:
-        print(f"Error loading safetensors file: {e}")
-        return {}, {}
+    if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
+        temp_safe_path = model_path + ".safetensors"
+        if not os.path.exists(temp_safe_path):
+            conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
+            if not conversion_successful:
+                return f"Conversion failed: {error_message}", ""
+        model_path = temp_safe_path
 
-class BaseMetaKeys:
-    RETURN_TYPES = ("STRING", "STRING")
-    RETURN_NAMES = ("metadata", "keys")
-    FUNCTION = "get_metakeys"
-    CATEGORY = "utils"
-    DESCRIPTION = "Loads a model and returns its metadata and layer names."
-
-    def _get_metakeys(self, model_name, model_type):
-        model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
-        model_weights = {}
-        metadata = {}
-        layer_shapes = ""
-
-        if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
-            temp_safe_path = model_path + ".safetensors"
-            if not os.path.exists(temp_safe_path):
-                conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
-                if not conversion_successful:
-                    return f"Conversion failed: {error_message}", ""
-            model_path = temp_safe_path
-
-        model_weights, metadata = load_metadata_from_safetensors(model_path)
-        metadata_str = str(metadata)
-        for layer_name, tensor in model_weights.items():
-            shape = str(tensor.shape)
-            shape = shape.replace("torch.Size(", "").replace(")", "")
-            name_shape = f"{layer_name}, {shape}\n"
-            layer_shapes += name_shape
-        return metadata_str, layer_shapes
+    model_weights, metadata = load_metadata_from_safetensors(model_path)
+    metadata_str = str(metadata)
+    for layer_name, tensor in model_weights.items():
+        shape = str(tensor.shape)
+        shape = shape.replace("torch.Size(", "").replace(")", "")
+        name_shape = f"{layer_name}, {shape}\n"
+        layer_shapes += name_shape
+    return metadata_str, layer_shapes
 
 
-
-class UNetMetaKeys(BaseMetaKeys):
+class ModelMetaKeys(io.ComfyNode):
     @classmethod
-    def INPUT_TYPES(s):
-        return {
-            "required": {
-                "unet_name": (folder_paths.get_filename_list("diffusion_models"), ),
-            }
-        }
+    def define_schema(cls):
+        return io.Schema(
+            node_id="ModelMetaKeys",
+            display_name="Get Diffusion Model Metadata & Keys",
+            category="ModelUtils/Keys",
+            description="Loads a diffusion model and returns its metadata and layer names.",
+            inputs=[
+                io.Combo.Input(
+                    "diffusionmodel_name",
+                    options=folder_paths.get_filename_list("diffusion_models"),
+                ),
+            ],
+            outputs=[
+                io.String.Output(display_name="metadata"),
+                io.String.Output(display_name="keys"),
+            ],
+        )
 
-    def get_metakeys(self, unet_name):
-        return self._get_metakeys(unet_name, "diffusion_models")
-
-class CLIPMetaKeys(BaseMetaKeys):
     @classmethod
-    def INPUT_TYPES(s):
-        return {
-            "required": {
-                "clip_name": (folder_paths.get_filename_list("text_encoders"), ),
-            }
-        }
+    def execute(cls, diffusionmodel_name: str) -> io.NodeOutput:
+        metadata, keys = _get_metakeys(diffusionmodel_name, "diffusion_models")
+        return io.NodeOutput(metadata, keys)
 
-    def get_metakeys(self, clip_name):
-        return self._get_metakeys(clip_name, "text_encoders")
 
-class LoRAMetaKeys(BaseMetaKeys):
+class TextEncoderMetaKeys(io.ComfyNode):
     @classmethod
-    def INPUT_TYPES(s):
-        return {
-            "required": {
-                "lora_name": (folder_paths.get_filename_list("loras"), ),
-            }
-        }
-    def get_metakeys(self, lora_name):
-        return self._get_metakeys(lora_name, "loras")
+    def define_schema(cls):
+        return io.Schema(
+            node_id="TextEncoderMetaKeys",
+            display_name="Get Text Encoder Metadata & Keys",
+            category="ModelUtils/Keys",
+            description="Loads a text encoder and returns its metadata and layer names.",
+            inputs=[
+                io.Combo.Input(
+                    "textencoder_name",
+                    options=folder_paths.get_filename_list("text_encoders"),
+                ),
+            ],
+            outputs=[
+                io.String.Output(display_name="metadata"),
+                io.String.Output(display_name="keys"),
+            ],
+        )
 
-class CheckpointMetaKeys(BaseMetaKeys):
     @classmethod
-    def INPUT_TYPES(s):
-        return {
-            "required": {
-                "ckpt_name": (folder_paths.get_filename_list("checkpoints"), ),
-            }
-        }
+    def execute(cls, textencoder_name: str) -> io.NodeOutput:
+        metadata, keys = _get_metakeys(textencoder_name, "text_encoders")
+        return io.NodeOutput(metadata, keys)
 
-    def get_metakeys(self, ckpt_name):
-        return self._get_metakeys(ckpt_name, "checkpoints")
 
-__all__ = ("UNetMetaKeys", "CLIPMetaKeys", "LoRAMetaKeys", "CheckpointMetaKeys")
+class LoRAMetaKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRAMetaKeys",
+            display_name="Get LoRA Metadata & Keys",
+            category="ModelUtils/Keys",
+            description="Loads a LoRA and returns its metadata and layer names.",
+            inputs=[
+                io.Combo.Input(
+                    "lora_name",
+                    options=folder_paths.get_filename_list("loras"),
+                ),
+            ],
+            outputs=[
+                io.String.Output(display_name="metadata"),
+                io.String.Output(display_name="keys"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, lora_name: str) -> io.NodeOutput:
+        metadata, keys = _get_metakeys(lora_name, "loras")
+        return io.NodeOutput(metadata, keys)
+
+
+class CheckpointMetaKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CheckpointMetaKeys",
+            display_name="Get Checkpoint Metadata & Keys",
+            category="ModelUtils/Keys",
+            description="Loads a checkpoint and returns its metadata and layer names.",
+            inputs=[
+                io.Combo.Input(
+                    "ckpt_name",
+                    options=folder_paths.get_filename_list("checkpoints"),
+                ),
+            ],
+            outputs=[
+                io.String.Output(display_name="metadata"),
+                io.String.Output(display_name="keys"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, ckpt_name: str) -> io.NodeOutput:
+        metadata, keys = _get_metakeys(ckpt_name, "checkpoints")
+        return io.NodeOutput(metadata, keys)
+
+
+class EmbeddingMetaKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="EmbeddingMetaKeys",
+            display_name="Get Embedding Metadata & Keys",
+            category="ModelUtils/Keys",
+            description="Loads an embedding and returns its metadata and layer names.",
+            inputs=[
+                io.Combo.Input(
+                    "embedding",
+                    options=folder_paths.get_filename_list("embeddings"),
+                ),
+            ],
+            outputs=[
+                io.String.Output(display_name="metadata"),
+                io.String.Output(display_name="keys"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, embedding: str) -> io.NodeOutput:
+        metadata, keys = _get_metakeys(embedding, "embeddings")
+        return io.NodeOutput(metadata, keys)

nodes/modelunloader.py

@@ -1,31 +0,0 @@
-from comfy.comfy_types import IO, ComfyNodeABC, InputTypeDict
-import comfy.model_management as mm
-
-class OffloadModelsNode(ComfyNodeABC):
-    @classmethod
-    def INPUT_TYPES(s) -> InputTypeDict:
-        return {
-            "required": {
-                "any": (IO.ANY, {}),
-                "force": (IO.BOOLEAN, {"default": False, "tooltip": "If True, forcibly unloads models from GPU."}),
-                "enable_offload": (IO.BOOLEAN, {"default": False, "tooltip": "If True, calls ComfyUI's `unload_all_models()` function."}),
-            }
-        }
-
-    RETURN_TYPES = (IO.ANY, )
-    RETURN_NAMES = ("any", )
-    FUNCTION = 'function'
-    CATEGORY = 'advanced'
-
-
-    def function(self, any, force, enable_offload):
-        if enable_offload == True:
-            print("OffloadModels: `enable_offload` is True. Calling `mm.unload_all_models()`.")
-            mm.unload_all_models()
-            if Force == True:
-                mm.soft_empty_cache(Force=True)
-            else:
-                mm.soft_empty_cache()
-        return (any,)
-
-__all__ = ("OffloadModelsNode")

nodes/prunekeys.py

@@ -0,0 +1,199 @@
+import os
+import re
+import folder_paths
+from comfy_api.latest import io
+from safetensors.torch import save_file
+from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
+
+
+def _prune_keys(model_name: str, model_type: str, keys_to_prune_str: str, 
+                use_regex: bool, output_filename: str) -> str:
+    """Shared logic for all PruneKeys nodes."""
+    model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
+
+    if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
+        temp_safe_path = model_path + ".safetensors"
+        if not os.path.exists(temp_safe_path):
+            conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
+            if not conversion_successful:
+                raise Exception(f"Conversion failed: {error_message}")
+        model_path_to_load = temp_safe_path
+    else:
+        model_path_to_load = model_path
+
+    model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
+
+    patterns = [p.strip() for p in keys_to_prune_str.strip().split('\n') if p.strip()]
+
+    if not patterns:
+        raise ValueError("No keys/patterns provided to prune.")
+
+    pruned_tensors = {}
+    for key, tensor in model_weights.items():
+        is_match = False
+        if use_regex:
+            if any(re.search(pattern, key) for pattern in patterns):
+                is_match = True
+        else:
+            if any(pattern in key for pattern in patterns):
+                is_match = True
+
+        if not is_match:
+            pruned_tensors[key] = tensor
+
+    model_dir = folder_paths.get_folder_paths(model_type)[0]
+    output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
+
+    save_file(pruned_tensors, output_path, metadata)
+
+    return output_path
+
+
+class ModelPruneKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="ModelPruneKeys",
+            display_name="Prune Diffusion Model Keys",
+            category="ModelUtils/Keys",
+            description="Loads a diffusion model, removes specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "diffusionmodel_name",
+                    options=folder_paths.get_filename_list("diffusion_models"),
+                ),
+                io.String.Input("keys_to_prune", multiline=True, default=""),
+                io.Boolean.Input("use_regex", default=False),
+                io.String.Input("output_filename", default="pruned_model"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, diffusionmodel_name: str, keys_to_prune: str, 
+                use_regex: bool, output_filename: str) -> io.NodeOutput:
+        path = _prune_keys(diffusionmodel_name, "diffusion_models", 
+                          keys_to_prune, use_regex, output_filename)
+        return io.NodeOutput(path)
+
+
+class TextEncoderPruneKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="TextEncoderPruneKeys",
+            display_name="Prune Text Encoder Keys",
+            category="ModelUtils/Keys",
+            description="Loads a text encoder, removes specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "textencoder_name",
+                    options=folder_paths.get_filename_list("text_encoders"),
+                ),
+                io.String.Input("keys_to_prune", multiline=True, default=""),
+                io.Boolean.Input("use_regex", default=False),
+                io.String.Input("output_filename", default="pruned_textencoder"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, textencoder_name: str, keys_to_prune: str, 
+                use_regex: bool, output_filename: str) -> io.NodeOutput:
+        path = _prune_keys(textencoder_name, "text_encoders", 
+                          keys_to_prune, use_regex, output_filename)
+        return io.NodeOutput(path)
+
+
+class LoRAPruneKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRAPruneKeys",
+            display_name="Prune LoRA Keys",
+            category="ModelUtils/Keys",
+            description="Loads a LoRA, removes specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "lora_name",
+                    options=folder_paths.get_filename_list("loras"),
+                ),
+                io.String.Input("keys_to_prune", multiline=True, default=""),
+                io.Boolean.Input("use_regex", default=False),
+                io.String.Input("output_filename", default="pruned_lora"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, lora_name: str, keys_to_prune: str, 
+                use_regex: bool, output_filename: str) -> io.NodeOutput:
+        path = _prune_keys(lora_name, "loras", 
+                          keys_to_prune, use_regex, output_filename)
+        return io.NodeOutput(path)
+
+
+class CheckpointPruneKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CheckpointPruneKeys",
+            display_name="Prune Checkpoint Keys",
+            category="ModelUtils/Keys",
+            description="Loads a checkpoint, removes specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "ckpt_name",
+                    options=folder_paths.get_filename_list("checkpoints"),
+                ),
+                io.String.Input("keys_to_prune", multiline=True, default=""),
+                io.Boolean.Input("use_regex", default=False),
+                io.String.Input("output_filename", default="pruned_checkpoint"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, ckpt_name: str, keys_to_prune: str, 
+                use_regex: bool, output_filename: str) -> io.NodeOutput:
+        path = _prune_keys(ckpt_name, "checkpoints", 
+                          keys_to_prune, use_regex, output_filename)
+        return io.NodeOutput(path)
+
+
+class EmbeddingPruneKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="EmbeddingPruneKeys",
+            display_name="Prune Embedding Keys",
+            category="ModelUtils/Keys",
+            description="Loads an embedding, removes specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "embedding",
+                    options=folder_paths.get_filename_list("embeddings"),
+                ),
+                io.String.Input("keys_to_prune", multiline=True, default=""),
+                io.Boolean.Input("use_regex", default=False),
+                io.String.Input("output_filename", default="pruned_embedding"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, embedding: str, keys_to_prune: str, 
+                use_regex: bool, output_filename: str) -> io.NodeOutput:
+        path = _prune_keys(embedding, "embeddings", 
+                          keys_to_prune, use_regex, output_filename)
+        return io.NodeOutput(path)

nodes/renamekeys.py

@@ -0,0 +1,194 @@
+import os
+import folder_paths
+from comfy_api.latest import io
+from safetensors.torch import save_file
+from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
+
+
+def _rename_keys(model_name: str, model_type: str, old_keys_str: str, 
+                 new_keys_str: str, output_filename: str) -> str:
+    """Shared logic for all RenameKeys nodes."""
+    model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
+
+    if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
+        temp_safe_path = model_path + ".safetensors"
+        if not os.path.exists(temp_safe_path):
+            conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
+            if not conversion_successful:
+                raise Exception(f"Conversion failed: {error_message}")
+        model_path_to_load = temp_safe_path
+    else:
+        model_path_to_load = model_path
+
+    model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
+
+    old_keys = [key.strip() for key in old_keys_str.strip().split('\n') if key.strip()]
+    new_keys = [key.strip() for key in new_keys_str.strip().split('\n') if key.strip()]
+
+    if len(old_keys) != len(new_keys):
+        raise ValueError("The number of old keys must match the number of new keys.")
+
+    key_map = dict(zip(old_keys, new_keys))
+
+    original_keys = list(model_weights.keys())
+    renamed_tensors = {}
+    for key in original_keys:
+        new_key = key_map.get(key, key)
+        renamed_tensors[new_key] = model_weights[key]
+
+    model_dir = folder_paths.get_folder_paths(model_type)[0]
+    output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
+
+    save_file(renamed_tensors, output_path, metadata)
+
+    return output_path
+
+
+class ModelRenameKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="ModelRenameKeys",
+            display_name="Rename Diffusion Model Keys",
+            category="ModelUtils/Keys",
+            description="Loads a diffusion model, renames specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "diffusionmodel_name",
+                    options=folder_paths.get_filename_list("diffusion_models"),
+                ),
+                io.String.Input("old_keys", multiline=True, default=""),
+                io.String.Input("new_keys", multiline=True, default=""),
+                io.String.Input("output_filename", default="renamed_model"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, diffusionmodel_name: str, old_keys: str, 
+                new_keys: str, output_filename: str) -> io.NodeOutput:
+        path = _rename_keys(diffusionmodel_name, "diffusion_models", 
+                           old_keys, new_keys, output_filename)
+        return io.NodeOutput(path)
+
+
+class TextEncoderRenameKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="TextEncoderRenameKeys",
+            display_name="Rename Text Encoder Keys",
+            category="ModelUtils/Keys",
+            description="Loads a text encoder, renames specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "textencoder_name",
+                    options=folder_paths.get_filename_list("text_encoders"),
+                ),
+                io.String.Input("old_keys", multiline=True, default=""),
+                io.String.Input("new_keys", multiline=True, default=""),
+                io.String.Input("output_filename", default="renamed_textencoder"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, textencoder_name: str, old_keys: str, 
+                new_keys: str, output_filename: str) -> io.NodeOutput:
+        path = _rename_keys(textencoder_name, "text_encoders", 
+                           old_keys, new_keys, output_filename)
+        return io.NodeOutput(path)
+
+
+class LoRARenameKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="LoRARenameKeys",
+            display_name="Rename LoRA Keys",
+            category="ModelUtils/Keys",
+            description="Loads a LoRA, renames specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "lora_name",
+                    options=folder_paths.get_filename_list("loras"),
+                ),
+                io.String.Input("old_keys", multiline=True, default=""),
+                io.String.Input("new_keys", multiline=True, default=""),
+                io.String.Input("output_filename", default="renamed_lora"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, lora_name: str, old_keys: str, 
+                new_keys: str, output_filename: str) -> io.NodeOutput:
+        path = _rename_keys(lora_name, "loras", 
+                           old_keys, new_keys, output_filename)
+        return io.NodeOutput(path)
+
+
+class CheckpointRenameKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="CheckpointRenameKeys",
+            display_name="Rename Checkpoint Keys",
+            category="ModelUtils/Keys",
+            description="Loads a checkpoint, renames specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "ckpt_name",
+                    options=folder_paths.get_filename_list("checkpoints"),
+                ),
+                io.String.Input("old_keys", multiline=True, default=""),
+                io.String.Input("new_keys", multiline=True, default=""),
+                io.String.Input("output_filename", default="renamed_checkpoint"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, ckpt_name: str, old_keys: str, 
+                new_keys: str, output_filename: str) -> io.NodeOutput:
+        path = _rename_keys(ckpt_name, "checkpoints", 
+                           old_keys, new_keys, output_filename)
+        return io.NodeOutput(path)
+
+
+class EmbeddingRenameKeys(io.ComfyNode):
+    @classmethod
+    def define_schema(cls):
+        return io.Schema(
+            node_id="EmbeddingRenameKeys",
+            display_name="Rename Embedding Keys",
+            category="ModelUtils/Keys",
+            description="Loads an embedding, renames specified keys, and saves it as a new safetensors file.",
+            inputs=[
+                io.Combo.Input(
+                    "embedding",
+                    options=folder_paths.get_filename_list("embeddings"),
+                ),
+                io.String.Input("old_keys", multiline=True, default=""),
+                io.String.Input("new_keys", multiline=True, default=""),
+                io.String.Input("output_filename", default="renamed_embedding"),
+            ],
+            outputs=[
+                io.String.Output(display_name="output_path"),
+            ],
+        )
+
+    @classmethod
+    def execute(cls, embedding: str, old_keys: str, 
+                new_keys: str, output_filename: str) -> io.NodeOutput:
+        path = _rename_keys(embedding, "embeddings", 
+                           old_keys, new_keys, output_filename)
+        return io.NodeOutput(path)

nodes/utils.py

@@ -0,0 +1,34 @@
+import torch
+from safetensors import safe_open
+from safetensors.torch import save_file
+from comfy.model_management import get_torch_device
+
+def convert_pt_to_safetensors(pt_path, safe_path):
+    try:
+        model = torch.load(pt_path, map_location="cpu")
+        if "state_dict" in model:
+            state_dict = model["state_dict"]
+        elif "model" in model:
+            state_dict = model["model"]
+        else:
+            state_dict = model
+        state_dict = {k.replace("state_dict.", ""): v for k, v in state_dict.items()}
+        metadata = {"format": "pt"}
+        save_file(state_dict, safe_path, metadata)
+        return True, ""
+    except Exception as e:
+        return False, str(e)
+
+def load_metadata_from_safetensors(safetensors_file: str) -> dict:
+    try:
+        with safe_open(safetensors_file, framework="pt", device="cpu") as f:
+            model_weights = {}
+            for key in f.keys():
+                model_weights[key] = f.get_tensor(key)
+            metadata = f.metadata()
+        if metadata is None:
+            metadata = {}
+        return model_weights, metadata
+    except Exception as e:
+        print(f"Error loading safetensors file: {e}")
+        return {}, {}

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "ComfyUI-ModelUtils"
 description = "[WIP]Custom nodes for handling, inspecting, modifying and creating various model files."
-version = "0.1.3"
+version = "0.1.7"
 license = { file = "LICENSE" }
 
 [project.urls]