""" Phase 2 — Model Composition & Merging ======================================= Implements: • Architecture introspection (DOM-tree-style layer parsing) • Union merging (SLERP, TIES, DARE-TIES, Task Arithmetic) • Intersection merging (Breadcrumbs / consensus filtering) • mergekit YAML generation (for mergekit-based merges) • Direct torch merging (for custom strategies) Usage: python -m phase2_merging.merge --strategy ties --models a/model1 b/model2 --output ./merged """ from __future__ import annotations import copy import math import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional import torch import yaml import typer from rich.console import Console from rich.tree import Tree from configs.settings import CFG, MERGES_DIR, HF_TOKEN from utils.logger import logger app = typer.Typer(help="Phase 2: Model merging") console = Console() # ───────────────────────────────────────────── # 1. Architecture Introspection # ───────────────────────────────────────────── @dataclass class LayerNode: """One node in the architecture tree.""" name: str kind: str # "embedding", "attention", "mlp", "norm", "head", "other" shape: tuple dtype: str numel: int children: list["LayerNode"] = field(default_factory=list) @property def params_m(self) -> float: return self.numel / 1e6 def __repr__(self): return f"LayerNode({self.name}, {self.kind}, shape={self.shape}, {self.params_m:.2f}M)" def _classify_layer(name: str) -> str: n = name.lower() if any(x in n for x in ["embed_tokens", "wte", "word_embeddings"]): return "embedding" if any(x in n for x in ["self_attn", "attention", "attn", "q_proj", "k_proj", "v_proj", "o_proj"]): return "attention" if any(x in n for x in ["mlp", "ffn", "feed_forward", "gate_proj", "up_proj", "down_proj", "fc1", "fc2"]): return "mlp" if any(x in n for x in ["norm", "layer_norm", "layernorm", "rmsnorm"]): return "norm" if any(x in n for x in ["lm_head", "embed_out", "output"]): return "head" return "other" def introspect_architecture(model_or_state_dict, model_id: str = "") -> LayerNode: """ Parse a model (or its state dict) into a DOM-tree-like LayerNode hierarchy. Works with nn.Module or a raw state dict. """ if hasattr(model_or_state_dict, "named_parameters"): state = {n: p.data for n, p in model_or_state_dict.named_parameters()} else: state = model_or_state_dict # already a state dict root = LayerNode(model_id or "model", "root", (), "", sum(p.numel() for p in state.values())) # Group by top-level prefix (e.g. "model.layers.0", "model.layers.1", ...) groups: dict[str, list[tuple[str, torch.Tensor]]] = {} for full_name, tensor in state.items(): top = full_name.split(".")[0] groups.setdefault(top, []).append((full_name, tensor)) for top_key, params in sorted(groups.items()): total_numel = sum(t.numel() for _, t in params) group_node = LayerNode( name = top_key, kind = _classify_layer(top_key), shape = (), dtype = "", numel = total_numel, ) for full_name, tensor in params: group_node.children.append(LayerNode( name = full_name, kind = _classify_layer(full_name), shape = tuple(tensor.shape), dtype = str(tensor.dtype).replace("torch.", ""), numel = tensor.numel(), )) root.children.append(group_node) return root def print_architecture_tree(root: LayerNode, max_depth: int = 3) -> None: """Render the LayerNode tree with rich.""" KIND_COLOR = { "embedding": "cyan", "attention": "magenta", "mlp": "green", "norm": "yellow", "head": "red", "other": "dim", "root": "bold white", } def _add(tree_node, layer: LayerNode, depth: int): if depth > max_depth: return color = KIND_COLOR.get(layer.kind, "white") label = (f"[{color}]{layer.name}[/{color}] " f"[dim]{layer.kind}[/dim] " f"[green]{layer.params_m:.2f}M[/green]") if layer.shape: label += f" [dim]{layer.shape}[/dim]" child_tree = tree_node.add(label) for child in layer.children: _add(child_tree, child, depth + 1) rich_tree = Tree(f"[bold]{root.name}[/bold] — {root.numel/1e9:.2f}B params") for child in root.children: _add(rich_tree, child, 1) console.print(rich_tree) def get_layer_groups(root: LayerNode) -> dict[str, list[str]]: """Return {kind: [param_names]} mapping for targeted merging.""" groups: dict[str, list[str]] = {} def _walk(node: LayerNode): if node.shape: # leaf groups.setdefault(node.kind, []).append(node.name) for c in node.children: _walk(c) _walk(root) return groups # ───────────────────────────────────────────── # 2. mergekit YAML generation # ───────────────────────────────────────────── def _mergekit_yaml(strategy: str, models: list[str], weights: Optional[list[float]] = None, density: float = 0.7, base_model: Optional[str] = None) -> dict: """Generate a mergekit-compatible merge config dict.""" if weights is None: weights = [1.0 / len(models)] * len(models) model_specs = [{"model": m, "parameters": {"weight": w}} for m, w in zip(models, weights)] if base_model: model_specs[0]["model"] = base_model # first slot = base for TIES/DARE cfg: dict[str, Any] = { "models": model_specs, "merge_method": strategy, "dtype": "bfloat16", "tokenizer_source": "union", } if strategy in ("ties", "dare_ties"): cfg["parameters"] = {"density": density, "normalize": True} elif strategy == "slerp": cfg["parameters"] = {"t": weights[1] if len(weights) > 1 else 0.5} elif strategy == "task_arithmetic": cfg["parameters"] = {"scaling_coefficient": 0.5} return cfg def run_mergekit( strategy: str, models: list[str], output_dir: Path, weights: Optional[list[float]] = None, density: float = 0.7, base_model: Optional[str] = None, ) -> Path: """ Write a mergekit YAML, then invoke mergekit-merge via subprocess. Requires: pip install mergekit """ output_dir.mkdir(parents=True, exist_ok=True) cfg_dict = _mergekit_yaml(strategy, models, weights, density, base_model) cfg_path = output_dir / "merge_config.yml" with open(cfg_path, "w") as f: yaml.dump(cfg_dict, f, default_flow_style=False) logger.info(f"[Merge] mergekit config → {cfg_path}") cmd = [ "mergekit-merge", str(cfg_path), str(output_dir), "--cuda", "--lazy-unpickle", ] if HF_TOKEN: cmd += ["--token", HF_TOKEN] logger.info(f"[Merge] Running: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: logger.error(f"mergekit failed:\n{result.stderr}") raise RuntimeError("mergekit-merge failed") logger.success(f"[Merge] Merged model saved → {output_dir}") return output_dir # ───────────────────────────────────────────── # 3. Direct torch merging (custom strategies) # ───────────────────────────────────────────── class TorchMerger: """ Pure-PyTorch implementations of: • SLERP — spherical linear interpolation • Task Arithmetic — delta-weight addition • TIES — trim, elect sign, merge • Breadcrumbs — intersection (consensus) merging All operate on loaded state dicts to avoid repeated model loading. """ # ── helpers ── @staticmethod def _load_state(model_id_or_path: str) -> dict[str, torch.Tensor]: from transformers import AutoModelForCausalLM, BitsAndBytesConfig import torch bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16) model = AutoModelForCausalLM.from_pretrained( model_id_or_path, quantization_config=bnb, device_map="cpu", # load to CPU for merging token=HF_TOKEN or None, trust_remote_code=True, ) return {k: v.clone().float() for k, v in model.state_dict().items()} @staticmethod def _slerp_tensors(t0: torch.Tensor, t1: torch.Tensor, alpha: float) -> torch.Tensor: """SLERP between two flat parameter vectors.""" orig_shape = t0.shape v0 = t0.flatten().double() v1 = t1.flatten().double() n0, n1 = v0.norm(), v1.norm() if n0 < 1e-8 or n1 < 1e-8: return (1 - alpha) * t0 + alpha * t1 u0, u1 = v0 / n0, v1 / n1 dot = torch.clamp((u0 * u1).sum(), -1.0, 1.0) theta = torch.acos(dot) if theta.abs() < 1e-6: return (1 - alpha) * t0 + alpha * t1 sin_theta = torch.sin(theta) result = (torch.sin((1 - alpha) * theta) / sin_theta * v0 + torch.sin(alpha * theta) / sin_theta * v1) return result.float().reshape(orig_shape) # ── public merge methods ── def slerp(self, model_a: str, model_b: str, alpha: float = 0.5) -> dict[str, torch.Tensor]: """SLERP: interpolate every matching parameter tensor.""" logger.info(f"[SLERP] Loading models...") sd_a = self._load_state(model_a) sd_b = self._load_state(model_b) merged = {} for key in sd_a: if key in sd_b and sd_a[key].shape == sd_b[key].shape: if sd_a[key].is_floating_point(): merged[key] = self._slerp_tensors(sd_a[key], sd_b[key], alpha) else: merged[key] = sd_a[key] # keep base for non-float (e.g. int8) else: merged[key] = sd_a[key] logger.success("[SLERP] Merge complete") return merged def task_arithmetic( self, base_model: str, models: list[str], scaling: float = 0.5, ) -> dict[str, torch.Tensor]: """ Task Arithmetic: merged = base + Σ scaling * (model_i - base) Union of capabilities. """ logger.info("[TaskArithmetic] Loading base...") base_sd = self._load_state(base_model) delta_sum: dict[str, torch.Tensor] = {} for mid in models: logger.info(f"[TaskArithmetic] Computing delta: {mid}") ft_sd = self._load_state(mid) for key in base_sd: if key in ft_sd and base_sd[key].shape == ft_sd[key].shape: delta = ft_sd[key].float() - base_sd[key].float() delta_sum[key] = delta_sum.get(key, torch.zeros_like(delta)) + delta merged = {} for key in base_sd: if key in delta_sum: merged[key] = base_sd[key].float() + scaling * delta_sum[key] else: merged[key] = base_sd[key] logger.success("[TaskArithmetic] Merge complete") return merged def ties( self, base_model: str, models: list[str], density: float = 0.7, scaling: float = 0.5, ) -> dict[str, torch.Tensor]: """ TIES (Trim, Elect Sign, Merge): 1. Compute deltas from base. 2. Trim lowest-magnitude changes per model (keep top `density` fraction). 3. Elect dominant sign per parameter. 4. Merge only parameters that agree with elected sign. """ logger.info("[TIES] Loading base...") base_sd = self._load_state(base_model) all_deltas: list[dict[str, torch.Tensor]] = [] for mid in models: logger.info(f"[TIES] Computing delta: {mid}") ft_sd = self._load_state(mid) delta = {} for key in base_sd: if key in ft_sd and base_sd[key].shape == ft_sd[key].shape: d = ft_sd[key].float() - base_sd[key].float() # Trim: zero out smallest (1 - density) fraction flat = d.abs().flatten() k = max(1, int(flat.numel() * density)) thresh = flat.kthvalue(flat.numel() - k + 1).values d[d.abs() < thresh] = 0.0 delta[key] = d all_deltas.append(delta) merged = {} for key in base_sd: stacked = torch.stack([ d[key] for d in all_deltas if key in d ], dim=0) # (n_models, *shape) # Elect sign by majority sign_sum = stacked.sign().sum(dim=0) elected = sign_sum.sign() # +1 / -1 / 0 elected[elected == 0] = 1 # break ties # Mask: keep only params that agree with elected sign mask = (stacked.sign() == elected.unsqueeze(0)).float() merged_delta = (stacked * mask).sum(dim=0) / (mask.sum(dim=0).clamp(min=1)) merged[key] = base_sd[key].float() + scaling * merged_delta logger.success("[TIES] Merge complete") return merged def breadcrumbs( self, base_model: str, models: list[str], density: float = 0.7, epsilon: float = 0.01, # consensus threshold ) -> dict[str, torch.Tensor]: """ Breadcrumbs (Intersection / Conservative merge): Only update parameters where ALL models agree on direction. Produces a safer, more conservative merged model. """ logger.info("[Breadcrumbs] Loading base...") base_sd = self._load_state(base_model) all_deltas: list[dict[str, torch.Tensor]] = [] for mid in models: logger.info(f"[Breadcrumbs] Computing delta: {mid}") ft_sd = self._load_state(mid) delta = {} for key in base_sd: if key in ft_sd and base_sd[key].shape == ft_sd[key].shape: delta[key] = ft_sd[key].float() - base_sd[key].float() all_deltas.append(delta) merged = {} for key in base_sd: deltas_for_key = [d[key] for d in all_deltas if key in d] if not deltas_for_key: merged[key] = base_sd[key] continue stacked = torch.stack(deltas_for_key, dim=0) # Consensus: all models must agree on sign (intersection) signs = stacked.sign() consensus = (signs == signs[0]).all(dim=0) # agree with first model avg_delta = stacked.mean(dim=0) avg_delta[~consensus] = 0.0 # zero out disagreements avg_delta[avg_delta.abs() < epsilon] = 0.0 # trim noise merged[key] = base_sd[key].float() + avg_delta logger.success("[Breadcrumbs] Merge complete") return merged def save(self, state_dict: dict[str, torch.Tensor], output_dir: Path, base_model_id: str) -> Path: """Save merged state dict as a HF model.""" from transformers import AutoTokenizer, AutoConfig, AutoModelForCausalLM output_dir.mkdir(parents=True, exist_ok=True) logger.info(f"[Merge] Saving to {output_dir} ...") # Load skeleton (config + tokenizer) from base, apply our weights config = AutoConfig.from_pretrained(base_model_id, token=HF_TOKEN or None, trust_remote_code=True) tok = AutoTokenizer.from_pretrained(base_model_id, token=HF_TOKEN or None, trust_remote_code=True) model = AutoModelForCausalLM.from_config(config) # Convert back to bf16 for storage merged_bf16 = {k: v.to(torch.bfloat16) for k, v in state_dict.items()} model.load_state_dict(merged_bf16, strict=False) model.save_pretrained(output_dir) tok.save_pretrained(output_dir) logger.success(f"[Merge] Model saved → {output_dir}") return output_dir # ───────────────────────────────────────────── # 4. High-level pipeline entry point # ───────────────────────────────────────────── def merge_models( strategy: str, models: list[str], output_dir: Optional[Path] = None, base_model: Optional[str] = None, density: float = 0.7, alpha: float = 0.5, use_mergekit: bool = True, ) -> Path: """ Unified entry point for all merge strategies. strategy ∈ {slerp, ties, dare_ties, task_arithmetic, breadcrumbs} use_mergekit=True → generates YAML and calls mergekit-merge (recommended for prod) use_mergekit=False → uses TorchMerger (no mergekit dependency needed) """ tag = "_".join(m.split("/")[-1] for m in models[:2]) out = output_dir or (MERGES_DIR / f"{strategy}_{tag}") if use_mergekit and strategy in ("slerp", "ties", "dare_ties", "task_arithmetic"): return run_mergekit(strategy, models, out, density=density, base_model=base_model) # Torch-native path merger = TorchMerger() base = base_model or models[0] if strategy == "slerp": sd = merger.slerp(models[0], models[1], alpha=alpha) elif strategy == "task_arithmetic": sd = merger.task_arithmetic(base, models[1:], scaling=alpha) elif strategy == "ties": sd = merger.ties(base, models[1:], density=density, scaling=alpha) elif strategy == "breadcrumbs": sd = merger.breadcrumbs(base, models[1:], density=density) else: raise ValueError(f"Unknown strategy: {strategy}") return merger.save(sd, out, base) # ───────────────────────────────────────────── # CLI # ───────────────────────────────────────────── @app.command() def run( strategy: str = typer.Argument(..., help="slerp|ties|dare_ties|task_arithmetic|breadcrumbs"), models: list[str] = typer.Option(..., "--model", "-m", help="Model IDs (repeat flag)"), output: Path = typer.Option(None, "--output", "-o"), base: str = typer.Option(None, "--base", "-b", help="Base model for delta strategies"), density: float = typer.Option(0.7, help="TIES/Breadcrumbs density"), alpha: float = typer.Option(0.5, help="SLERP/TaskArith interpolation weight"), torch_only: bool = typer.Option(False, "--torch-only", help="Skip mergekit, use TorchMerger"), introspect: str = typer.Option(None, "--introspect", help="Print architecture tree for a model ID"), ): if introspect: from transformers import AutoModelForCausalLM logger.info(f"Loading {introspect} for introspection...") m = AutoModelForCausalLM.from_pretrained(introspect, device_map="cpu", token=HF_TOKEN or None) root = introspect_architecture(m, introspect) print_architecture_tree(root) return out = merge_models( strategy=strategy, models=models, output_dir=output, base_model=base, density=density, alpha=alpha, use_mergekit=not torch_only, ) console.print(f"[green]✓ Merged model ready at: {out}[/green]") if __name__ == "__main__": app()