1"""
2Tensor surgery: Graft FP8 attention projections into the MXFP4 model.
3
4Takes the original MXFP4 model (experts as _blocks/_scales) and replaces
5the bf16 attention projection weights with FP8 e4m3fn weights + scales
6from the llmcompressor-calibrated model.
7
8Result: MXFP4 experts + FP8 static attention, single compressed-tensors config.
9"""
10
11import os
12import json
13import shutil
14import logging
15
16logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
17log = logging.getLogger(__name__)
18
19import torch
20from safetensors import safe_open
21from safetensors.torch import save_file
22
23MXFP4_DIR = "/path/to/gpt-oss-120b-mxfp4"
24FP8_DIR = "/path/to/gpt-oss-120b-mxfp4-fp8attn"
25OUTPUT_DIR = "/path/to/gpt-oss-120b-mxfp4-fp8attn-v2"
26
27ATTN_PROJS = ("self_attn.q_proj", "self_attn.k_proj", "self_attn.v_proj", "self_attn.o_proj")
28SKIP_DIRS = ("original", "metal")
29
30# Step 1: Collect FP8 attention tensors from the calibrated model
31log.info("Loading FP8 attention tensors from %s", FP8_DIR)
32with open(os.path.join(FP8_DIR, "model.safetensors.index.json")) as f:
33 fp8_index = json.load(f)
34
35fp8_tensors = {}
36fp8_shards_loaded = set()
37
38for key, shard in fp8_index["weight_map"].items():
39 if not any(proj in key for proj in ATTN_PROJS):
40 continue
41 if shard not in fp8_shards_loaded:
42 shard_path = os.path.join(FP8_DIR, shard)
43 with safe_open(shard_path, framework="pt") as sf:
44 for k in sf.keys():
45 if any(proj in k for proj in ATTN_PROJS):
46 fp8_tensors[k] = sf.get_tensor(k)
47 fp8_shards_loaded.add(shard)
48
49log.info("Loaded %d FP8 attention tensors (weights + scales)", len(fp8_tensors))
50
51# Step 2: Create output dir, copy non-safetensor files
52os.makedirs(OUTPUT_DIR, exist_ok=True)
53
54for fname in os.listdir(MXFP4_DIR):
55 if fname in SKIP_DIRS:
56 continue
57 src = os.path.join(MXFP4_DIR, fname)
58 if os.path.isfile(src) and not fname.endswith(".safetensors"):
59 shutil.copy2(src, os.path.join(OUTPUT_DIR, fname))
60 log.info("Copied %s", fname)
61
62# Step 3: Process each MXFP4 shard
63with open(os.path.join(MXFP4_DIR, "model.safetensors.index.json")) as f:
64 mxfp4_index = json.load(f)
65
66shard_keys = {}
67for key, shard in mxfp4_index["weight_map"].items():
68 shard_keys.setdefault(shard, []).append(key)
69
70new_weight_map = {}
71
72for shard_name, keys in sorted(shard_keys.items()):
73 src_path = os.path.join(MXFP4_DIR, shard_name)
74 dst_path = os.path.join(OUTPUT_DIR, shard_name)
75
76 has_attn = any(any(proj in k for proj in ATTN_PROJS) for k in keys)
77
78 if not has_attn:
79 shutil.copy2(src_path, dst_path)
80 for k in keys:
81 new_weight_map[k] = shard_name
82 log.info("Copied shard %s (no attn weights)", shard_name)
83 continue
84
85 tensors = {}
86 with safe_open(src_path, framework="pt") as sf:
87 for k in sf.keys():
88 tensors[k] = sf.get_tensor(k)
89
90 replaced = 0
91 added = 0
92
93 for k in list(tensors.keys()):
94 if k in fp8_tensors:
95 tensors[k] = fp8_tensors[k]
96 replaced += 1
97 if k.endswith(".weight") and any(proj in k for proj in ATTN_PROJS):
98 base = k.removesuffix(".weight")
99 for suffix in (".weight_scale", ".input_scale"):
100 scale_key = base + suffix
101 if scale_key in fp8_tensors and scale_key not in tensors:
102 tensors[scale_key] = fp8_tensors[scale_key]
103 added += 1
104
105 save_file(tensors, dst_path)
106 for k in tensors:
107 new_weight_map[k] = shard_name
108 log.info("Wrote shard %s (replaced %d, added %d scale tensors)", shard_name, replaced, added)
109
110# Step 4: Write updated safetensors index
111new_index = {
112 "metadata": mxfp4_index.get("metadata", {}),
113 "weight_map": dict(sorted(new_weight_map.items())),
114}
115with open(os.path.join(OUTPUT_DIR, "model.safetensors.index.json"), "w") as f:
116 json.dump(new_index, f, indent=2)
117log.info("Wrote safetensors index (%d keys)", len(new_weight_map))
118
119# Step 5: Write config.json with dual quantization config
120with open(os.path.join(MXFP4_DIR, "config.json")) as f:
121 config = json.load(f)
122
123config["quantization_config"] = {
124 "modules_to_not_convert": [
125 "model.layers.*.mlp.router",
126 "model.embed_tokens",
127 "lm_head"
128 ],
129 "quant_method": "mxfp4"
130}
131config["attn_quantization_config"] = {
132 "quant_method": "fp8",
133 "activation_scheme": "static"
134}
135
136with open(os.path.join(OUTPUT_DIR, "config.json"), "w") as f:
137 json.dump(config, f, indent=2)
138log.info("Wrote config.json with dual MXFP4+FP8 quantization config")