#!/usr/bin/env python3 """ASHQ1-MMPROJ — Non-uniform quantization for CLIP/mmproj GGUF files.""" import argparse,os,sys import numpy as np import gguf from gguf.constants import GGMLQuantizationType as QT BLOCK_ALIGN={ QT.Q4_K:256,QT.Q5_K:256,QT.Q6_K:256, QT.Q8_0:32,QT.Q5_0:32,QT.Q4_0:32, QT.F16:1,QT.F32:1,QT.BF16:1, } QT_BLOCK={ QT.Q4_K:(256,144),QT.Q5_K:(256,176),QT.Q6_K:(256,210), QT.Q8_0:(32,34),QT.Q5_0:(32,22),QT.Q4_0:(32,18), } TIER_TO_QT={ "F32":QT.F32,"F16":QT.F16,"BF16":QT.BF16, "Q8_0":QT.Q8_0,"Q6_K":QT.Q6_K,"Q5_K":QT.Q5_K,"Q4_K":QT.Q4_K, } BPW={ "F32":32.0,"F16":16.0,"BF16":16.0, "Q8_0":8.5,"Q6_K":6.5625,"Q5_K":5.5,"Q4_K":4.5, } CRITICAL=( "patch_embd","position_embd","mm.0","mm.1","mm.2", "pre_ln","post_ln","ln1","ln2","ln_q","norm", ) PROFILES={ "balanced":{ "attn_qkv":"Q6_K","attn_out":"Q6_K", "ffn_up":"Q6_K","ffn_down":"Q6_K", "critical":"F32","default":"Q8_0", }, "compact":{ "attn_qkv":"Q5_K","attn_out":"Q5_K", "ffn_up":"Q5_K","ffn_down":"Q5_K", "critical":"F32","default":"Q8_0", }, "fidelity":{ "attn_qkv":"Q8_0","attn_out":"Q8_0", "ffn_up":"Q8_0","ffn_down":"Q8_0", "critical":"F32","default":"F16", }, } UPGRADE_CHAIN={"Q4_K":"Q5_K","Q5_K":"Q6_K","Q6_K":"Q8_0","Q8_0":"F16"} FILE_TYPE_NAME={ "F32":"ALL_F32","F16":"MOSTLY_F16","BF16":"MOSTLY_BF16", "Q8_0":"MOSTLY_Q8_0","Q6_K":"MOSTLY_Q6_K","Q5_K":"MOSTLY_Q5_K_M","Q4_K":"MOSTLY_Q4_K_M", } def is_critical(name:str)->bool: low=name.lower() return any(c in low for c in CRITICAL) def tensor_role(name:str)->str: parts=name.split(".") if len(parts)>=4 and parts[0]=="v" and parts[1]=="blk": return parts[3] return name def block_index(name:str): parts=name.split(".") if len(parts)>=3 and parts[0]=="v" and parts[1]=="blk": try:return int(parts[2]) except ValueError:return None return None def row_length(shape)->int: return int(shape[0]) if len(shape)>0 else 1 def best_quant_fit(shape,tier:str)->str: cur=tier seen=set() while cur not in seen: seen.add(cur) if fits_alignment(shape,cur):return cur nxt=UPGRADE_CHAIN.get(cur) if nxt is None:break cur=nxt return "F16" def fits_alignment(shape,tier:str)->bool: qt=TIER_TO_QT[tier] align=BLOCK_ALIGN.get(qt,1) if align<=1:return True return row_length(shape) % align==0 def pick_tier(name:str,shape,n_blocks:int,profile:dict,deep_boost:int)->str: if is_critical(name):return profile["critical"] # Keep 1D vectors in F32 (layer scale, bias) if len(shape)<2 or int(np.prod(shape))<4096:return "F32" role=tensor_role(name) tier=profile.get(role,profile["default"]) idx=block_index(name) if idx is not None and deep_boost>0 and idx>=n_blocks-deep_boost: tier=UPGRADE_CHAIN.get(tier,tier) tier=best_quant_fit(shape,tier) return "F16" if tier=="BF16" else tier def _decode(t)->np.ndarray: raw=np.asarray(t.data) name=t.tensor_type.name nshape=tuple(int(d) for d in reversed(t.shape)) flat=raw.reshape(-1) if name=="BF16": u=flat.view(np.uint16).astype(np.uint32)<<16 return u.view(np.float32).reshape(nshape) if name=="F16": return flat.view(np.float16).astype(np.float32).reshape(nshape) if name=="F32": return flat.view(np.float32).reshape(nshape) return gguf.quants.dequantize(raw,t.tensor_type).astype(np.float32).reshape(nshape) def quantize_mmproj(src:str,dst:str,profile_name:str,deep_boost:int,dry_run:bool)->bool: profile=PROFILES[profile_name] reader=gguf.GGUFReader(src) arch=None for key,field in reader.fields.items(): if key=="general.architecture": arch=str(bytes(field.parts[field.data[0]]),"utf-8") break if arch!="clip": print(f" [WARN] architecture='{arch}' (clip expected)") n_blocks=0 for t in reader.tensors: idx=block_index(t.name) if idx is not None:n_blocks=max(n_blocks,idx+1) print(f" ViT blocks detected: {n_blocks}") plan=[] src_bits=0.0 dst_bits=0.0 for t in reader.tensors: shape=list(t.shape) n_el=int(np.prod(shape)) tier=pick_tier(t.name,shape,n_blocks,profile,deep_boost) cur_name=t.tensor_type.name if cur_name in BPW and BPW[cur_name]4} tensor(s) ({mib:>8.1f} MiB)") print(f"\n Source (tensors) : {src_mib:>8.1f} MiB") print(f" Target (tensors) : {dst_mib:>8.1f} MiB") print(f" Reduction : {(1-dst_mib/src_mib)*100:>7.1f}%") heavy=[(t.name,tier,n*BPW[tier]/8/1024/1024,list(t.shape)) for t,tier,n in plan if BPW[tier]>=16.0 and n*BPW[tier]/8/1024/1024>1.0] if heavy: heavy.sort(key=lambda x:-x[2]) print(f"\n High-precision tensors kept in 16/32-bit ({len(heavy)}):") for name,tier,mib,shape in heavy[:20]: print(f" {name[:46]:46s} {tier:<5} {mib:>7.1f} MiB shape={shape}") if dry_run: print("\n Dry run complete — no file written.") return True writer=gguf.GGUFWriter(dst,arch or "clip") skip={"GGUF.version","GGUF.tensor_count","GGUF.kv_count","general.architecture","general.quantization_version","general.file_type"} for key,field in reader.fields.items(): if key in skip:continue try:_copy_field(writer,key,field) except Exception as e:print(f" [WARN] KV '{key}' omitted: {e}") writer.add_quantization_version(2) dominant_tier=max(by_tier,key=lambda t:by_tier[t][1]) dominant_name=FILE_TYPE_NAME.get(dominant_tier,"MOSTLY_Q8_0") writer.add_file_type(getattr(gguf.LlamaFileType,dominant_name)) for t,tier,_ in plan: data=_decode(t) qt=TIER_TO_QT[tier] if tier=="F32": out=np.ascontiguousarray(data.astype(np.float32)) writer.add_tensor(t.name,out) continue if tier=="F16": out=np.ascontiguousarray(data.astype(np.float16)) writer.add_tensor(t.name,out) continue if tier=="BF16": out=data.astype(np.float16) writer.add_tensor(t.name,out,raw_dtype=QT.F16) continue try: out=gguf.quants.quantize(data.astype(np.float32),qt) except NotImplementedError: qt=TIER_TO_QT["Q8_0"] out=gguf.quants.quantize(data.astype(np.float32),qt) blk,bsz=QT_BLOCK[qt] nshape=list(data.shape) nshape[-1]=(nshape[-1]//blk)*bsz writer.add_tensor(t.name,out,raw_dtype=qt,raw_shape=nshape) writer.write_header_to_file() writer.write_kv_data_to_file() writer.write_tensors_to_file() writer.close() real=os.path.getsize(dst)/1024/1024 print(f"\n Output: {dst}") print(f" Actual size: {real:.1f} MiB") drift=real/dst_mib if dst_mib>0 else 0 if drift>1.05: print(f" ⚠ Size divergence ×{drift:.2f}") else: print(f" ✓ Size matches plan (×{drift:.2f})") return True def _copy_field(writer,key:str,field): from gguf.constants import GGUFValueType as VT vtype=field.types[0] if vtype==VT.ARRAY: itype=field.types[1] if itype==VT.STRING: vals=[str(bytes(field.parts[i]),"utf-8") for i in field.data] else: vals=[field.parts[i].tolist()[0] for i in field.data] writer.add_array(key,vals) return if vtype==VT.STRING: writer.add_string(key,str(bytes(field.parts[field.data[0]]),"utf-8")) return val=field.parts[field.data[0]].tolist()[0] emit={ VT.UINT8:writer.add_uint8,VT.INT8:writer.add_int8, VT.UINT16:writer.add_uint16,VT.INT16:writer.add_int16, VT.UINT32:writer.add_uint32,VT.INT32:writer.add_int32, VT.UINT64:writer.add_uint64,VT.INT64:writer.add_int64, VT.FLOAT32:writer.add_float32,VT.FLOAT64:writer.add_float64, VT.BOOL:writer.add_bool, } fn=emit.get(vtype) if fn is None:raise ValueError(f"Unhandled value type {vtype}") fn(key,val) def main(args_list=None): p=argparse.ArgumentParser(description="ASHQ1-MMPROJ — Non-uniform quantization for mmproj/CLIP") p.add_argument("--model",required=True,help="Source mmproj GGUF (BF16/F16)") p.add_argument("--output",default=None,help="Output path") p.add_argument("--profile",choices=list(PROFILES),default="balanced") p.add_argument("--deep-boost",type=int,default=3,help="Number of trailing blocks boosted one tier") p.add_argument("--dry-run",action="store_true",help="Show quantization plan without writing") a=p.parse_args(args_list) if not os.path.isfile(a.model): print(f"ERROR: file not found: {a.model}") return 1 if not a.output: base=os.path.splitext(os.path.basename(a.model))[0] mm_suffix="F16" if "-F16" in base or "-f16" in base else "BF16" a.output=f"mmproj-{mm_suffix}-to-ASHQ1-{a.profile.capitalize()}.gguf" print("=== ASHQ1-MMPROJ ===") print(f"Source : {a.model}") print(f"Profile : {a.profile} (deep-boost={a.deep_boost})") print(f"Output : {a.output}\n") ok=quantize_mmproj(a.model,a.output,a.profile,a.deep_boost,a.dry_run) return 0 if ok else 1 if __name__=="__main__": sys.exit(main())