| |
| """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"] |
| |
| 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]<BPW[tier]: |
| tier=cur_name |
| if cur_name=="BF16" and tier=="BF16": |
| tier="F16" |
| plan.append((t,tier,n_el)) |
| src_bits+=n_el*BPW.get(cur_name,32.0) |
| dst_bits+=n_el*BPW[tier] |
| src_mib=src_bits/8/1024/1024 |
| dst_mib=dst_bits/8/1024/1024 |
| by_tier={} |
| for _,tier,n_el in plan: |
| e=by_tier.setdefault(tier,[0,0.0]) |
| e[0]+=1 |
| e[1]+=n_el*BPW[tier]/8/1024/1024 |
| print("\n Tier distribution:") |
| for tier in sorted(by_tier,key=lambda x:-BPW[x]): |
| cnt,mib=by_tier[tier] |
| print(f" {tier:<5}: {cnt:>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()) |
|
|
