File size: 58,426 Bytes
90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 271bd15 90f10d2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 | """Cast math: orient, fuse, quantize, pack writes."""
from pathlib import Path
import os
import struct
import sys
import time
from vlib import ctx
from vlib.ui import _confirm, _fail, _noninteractive, _note, _ok, _progress_pref, _say, _step, _transform, _warn
from vlib.tensors import _gguf_raw_to_f32, _is_delta_safetensors, _is_ggml_quant, _is_gguf_file, _safetensors_write_repl, decode_to_f32, encode_from_f32, read_safetensors, rewrite_safetensors_tensor
from vlib.sources import OUTPUT_ALIASES, _nested, detect_output_tensor, largest_2d, open_source
from vlib.registry import log_operation
def _load_voice(voice_path):
"""Load a voice artifact. Returns (f32_array_or_None, ref, raw_bytes_or_None, ggml_type_or_None).
Quantized GGUF artifacts carry raw blocks; safetensors artifacts decode to f32."""
src = open_source(str(voice_path))
names = [n for n in src.names() if n != "delta.voice.marker"]
if not names:
_fail(f" ✗ No tensors found in {voice_path}.")
tname = names[0] if len(names) == 1 else "output.weight" if "output.weight" in names else names[0]
ref = src.ref(tname)
if src.kind == "gguf" and _is_ggml_quant(ref.dtype):
return None, ref, src.read_raw(tname), ref.dtype
return src.read_f32(tname), ref, None, None
def _orient(arr, shape, target_shape):
import numpy as np
if tuple(shape) == tuple(target_shape):
return arr
if tuple(shape) == tuple(target_shape[::-1]):
if ctx.VERBOSE:
_note("Transposing to match the target layout…")
return np.ascontiguousarray(arr.T)
return None
def _cast_st_prepare(target, tensors, vf32, vref, meta, args, vraw=None, vqtype=None, strict_target=False, target_name=None):
"""Per-tensor compute for a safetensors target. Returns (tname, data, tdtype, tshape).
target_name (from _resolve_pack_targets) overrides lookup; None keeps the
original single-tensor behavior byte-for-byte."""
import numpy as np
if vraw is not None:
# quant artifact -> safetensors target: dequant is the only requant path
from gguf.constants import GGMLQuantizationType as Q
from gguf.quants import quant_shape_to_byte_shape
vf32 = _gguf_raw_to_f32(vraw, Q[vqtype], tuple(reversed(tuple(int(x) for x in vref.shape))))
names = list(tensors.keys())
if target_name is not None:
tname = target_name if target_name in names else None
if tname is None:
_fail(f" ✗ Resolved target '{target_name}' for '{vref.name}' not in {target}.")
else:
tname = vref.name if vref.name in names else None
if tname is None and strict_target:
_fail(f" ✗ Tensor '{vref.name}' not found in target {target}. Voicepack requires exact match (not mapping to another tensor).")
if tname is None:
tname = detect_output_tensor(names, None) or largest_2d(names, {n: tensors[n]["shape"] for n in names})
if tname is None:
_fail(" ✗ Could not find a matching tensor in the target.")
tshape = tuple(tensors[tname]["shape"])
tdtype = tensors[tname]["dtype"]
# --quant override for safetensors target: only F32/F16/BF16/keep/target make sense (no GGUF quants).
_qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
if _qo is None:
_fail(f" ✗ Invalid --quant '{getattr(args, 'quant', None)}' for safetensors target. Use keep|target|f32|f16|bf16.")
if _qo not in ("keep", "target", "f32", "f16", "bf16"):
_fail(f" ✗ --quant {_qo} needs GGUF target (safetensors has no quant {_qo}). Use keep|target|f32|f16|bf16.")
if _qo in ("f32", "f16", "bf16"):
_want = {"f32": "F32", "f16": "F16", "bf16": "BF16"}[_qo]
if _want != tdtype.upper():
from vlib.tensors import dtype_bytes as _db
try:
_same_size = _db(_want) == _db(tdtype.upper())
except Exception:
_same_size = False
if not _same_size:
_fail(f" ✗ --quant {_qo} would change '{tname}' {tdtype} → {_want} (different byte size; safetensors casts keep byte size). Use --quant keep|target.")
_warn(f" Re-encoding target {tdtype} → {_want} for --quant {_qo} (extra conversion).")
tdtype = _want
# Show quant/target by default for tensor commands (advanced) and verbose for voice.
try:
_svq = vqtype if vraw is not None else vref.dtype
except Exception:
_svq = "?"
if ctx.VERBOSE:
_note(f"Voice '{vref.name}' is {_svq} {tuple(vref.shape)} → target '{tname}' is {tdtype} {tshape}")
if vraw is not None and ctx.VERBOSE:
_note(f"Dequanting {vqtype} once to F32 for safetensors target (unavoidable, safetensors has no quant).")
oriented = _orient(vf32, vref.shape, tshape)
if oriented is None:
_fail(f" ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {tshape}.")
if (meta or {}).get("type") == "delta":
_step(f"Fusing style into {tname} (delta + head → {tdtype}, kept)…")
with open(target, "rb") as f:
_hs = struct.unpack("<Q", f.read(8))[0]
info = tensors[tname]
with open(target, "rb") as f:
f.seek(8 + _hs + info["data_offsets"][0])
head_f32 = decode_to_f32(f.read(info["data_offsets"][1] - info["data_offsets"][0]), tdtype, tshape)
oriented = np.ascontiguousarray(head_f32 + oriented)
data = encode_from_f32(oriented, tdtype)
return tname, data, tdtype, tshape
def _cast_core(target, target_format, out_path, vf32, vref, meta, args, op="cast", vraw=None, vqtype=None, strict_target=False, target_name=None):
if getattr(args, "speak", False) and target_format != "gguf":
_fail(" ✗ --speak needs a .gguf target. Re-run without --speak for .safetensors, or point cast at a .gguf file.")
if target_format == "gguf":
_cast_gguf(target, out_path, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype, strict_target=strict_target, target_name=target_name)
return
# safetensors target
header, tensors = read_safetensors(target)
t0s = time.time()
tname, data, tdtype, tshape = _cast_st_prepare(target, tensors, vf32, vref, meta, args,
vraw=vraw, vqtype=vqtype, strict_target=strict_target,
target_name=target_name)
_step(f"Replacing {tname} ({tdtype} {tshape})…")
rewrite_safetensors_tensor(target, out_path, tname, data, tdtype, tshape)
file_size = os.path.getsize(out_path)
try:
_mb = len(data) / 1e6
_rate = f" @ {(_mb) / max(time.time() - t0s, 1e-6):.0f} MB/s" if _mb >= 0.1 else ""
_note(f"Kept '{tname}': {_mb:.1f} MB → {tdtype}{_rate}")
except Exception:
pass
log_operation({"op": op, "target": target, "format": "safetensors", "tensor": tname,
"dtype": tdtype, "shape": list(tshape), "output": out_path, "status": "ok"})
_ok(f"Done → {out_path}")
_say(f" Saved → {out_path} ({tdtype} {list(tshape)})")
def _is_head_name(name):
if not name:
return False
try:
if name in OUTPUT_ALIASES or name in ("token_embd.weight",):
return True
for a in OUTPUT_ALIASES:
if name.endswith(a):
return True
except Exception:
pass
return False
def _parse_quant_override(s):
# Returns normalized key: keep|target or Q-type name upper (e.g. Q4_0). Trivial: accept all GGML quants.
if s is None:
return "keep"
k = str(s).strip().lower().replace("-", "_")
if k in ("keep", "target"):
return k
# Aliases for float family.
if k in ("fp32",):
k = "f32"
if k in ("fp16",):
k = "f16"
# Try any GGMLQuantizationType name.
try:
from gguf.constants import GGMLQuantizationType as _Qp
# Case-insensitive match against all Q names.
for _q in _Qp:
if _q.name.lower() == k:
return k
except Exception:
pass
# Fallback known list (when gguf import fails).
_known = {"f32", "f16", "bf16", "q8_0", "q8_1", "q8_k", "q1_0", "q4_0", "q4_1", "q5_0", "q5_1",
"q2_k", "q3_k", "q4_k", "q5_k", "q6_k", "iq1_s", "iq1_m", "iq2_xxs", "iq2_xs", "iq2_s",
"iq3_xxs", "iq3_s", "iq4_nl", "iq4_xs", "tq1_0", "tq2_0", "mxfp4", "nvfp4"}
if k in _known:
return k
return None
def _quant_from_key(key):
# key normalized via _parse_quant_override (lower). Returns Q type or None for keep/target.
if key in ("keep", "target", None):
return None
try:
from gguf.constants import GGMLQuantizationType as _Qq
for _q in _Qq:
if _q.name.lower() == key:
return _q
# Upper direct (e.g. Q4_0).
return _Qq[key.upper()]
except Exception:
return None
def _qerr(e):
"""Readable lib error: gguf often raises bare NotImplementedError ('')."""
s = str(e).strip()
return f"{type(e).__name__}: {s}" if s else type(e).__name__
_NEAREST_CACHE = {}
def _nearest_encodable(target_q):
"""Nearest gguf-encodable quant to target_q by bits/param (ties → larger =
quality-preserving). The lib decodes everything but encodes only a subset
(no K/IQ quants); this keeps fused tensors near the original size instead
of ballooning to F16. Support probed once per process (version-proof)."""
import numpy as np
from gguf.constants import GGMLQuantizationType as Q
from gguf.constants import GGML_QUANT_SIZES as S
from gguf.quants import quantize
key = getattr(target_q, "name", str(target_q))
if key in _NEAREST_CACHE:
return _NEAREST_CACHE[key]
tbpp = S[target_q][1] * 8 / S[target_q][0]
probe = (np.arange(64 * 64, dtype=np.float32).reshape(64, 64) - 2048) / 64
scored = []
for name in ("Q8_0", "Q5_1", "Q5_0", "Q4_1", "Q4_0", "MXFP4", "F16"):
cand = getattr(Q, name, None)
if cand is None or cand == target_q:
continue
try:
quantize(probe, cand)
except Exception:
continue
cbpp = S[cand][1] * 8 / S[cand][0]
scored.append((abs(cbpp - tbpp), -cbpp, cand))
best = min(scored)[2] if scored else None
_NEAREST_CACHE[key] = best
return best
def _quantize_to_Q(fused, req_q):
# Quantize F32 fused array to any Q (F32/F16/BF16 via astype, else gguf quantize).
# Returns (q_head, out_qtype). Raises on unsupported (e.g. Q1_0/Q8_1/Q8_K NotImplemented).
import numpy as np
from gguf.constants import GGMLQuantizationType as Q
from gguf.quants import quantize
if req_q in (Q.F32,):
return fused.astype(np.float32), Q.F32
if req_q in (Q.F16,):
return fused.astype(np.float16), Q.F16
try:
_b = getattr(Q, "BF16", None)
except Exception:
_b = None
if _b is not None and req_q == _b:
_u32 = fused.astype(np.float32).view(np.uint32)
return (((_u32 + 0x7FFF + ((_u32 >> 16) & 1)) >> 16).astype(np.uint16), _b)
# All other GGML quants (Q4_0/Q4_K/Q6_K/IQ*/TQ*/MXFP4/...) via single quantize call (trivial).
return quantize(fused, req_q), req_q
def _fuse_delta(target_tensor, delta_f32, strict=False):
"""Dequant the target head once to F32 and fuse the delta (head + delta).
Single seam for the three dequant→fuse sites in _gguf_fuse_one.
strict (safetensors delta path only): fail clean on shape mismatch, exactly as before."""
import numpy as np
from gguf.constants import GGMLQuantizationType as Q
if target_tensor.tensor_type in (Q.F32, Q.F16):
try:
head_f32 = target_tensor.data.astype(np.float32)
except Exception:
head_f32 = np.asarray(target_tensor.data, dtype=np.float32)
else:
from gguf.quants import dequantize as _deq
head_f32 = _deq(target_tensor.data, target_tensor.tensor_type)
if strict and head_f32.shape != delta_f32.shape:
sys.stdout.write("\n")
_fail(f" ✗ Style shape {delta_f32.shape} doesn't match model head {head_f32.shape}.")
return np.ascontiguousarray(head_f32 + delta_f32)
def _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=None, vqtype=None):
"""Fuse one voice tensor into its target GGUF tensor form. Returns (q_head, out_qtype)."""
import numpy as np
from gguf.constants import GGMLQuantizationType as Q
from gguf.quants import quantize
# Philosophy: preserve quality, avoid dequant->quant cycles.
# - GGUF quant voice (vraw) is kept at its obtained quant (never upscale Q2->Q6, no inventing data).
# - Safetensors high-fidelity lm_head (voice) is quantized once to Q8_0.
# - Generic tensors keep target quant (F32 norms stay F32, never forced Q8_0).
# - All GGML quants supported (trivial passthrough + quantize where implemented).
_quant_override = _parse_quant_override(getattr(args, "quant", None) or "keep")
if _quant_override is None:
_fail(f" ✗ Invalid --quant '{getattr(args, 'quant', None)}'. Use keep|target|<any quant like q8_0,q4_0,q4_k,q6_k,iq4_nl,f16,f32,bf16>.")
_is_head = _is_head_name(vref.name) or _is_head_name(tname)
_is_delta = (meta or {}).get("type") == "delta"
# Show quant/target by default for advanced tensor commands (and verbose for voice)
try:
_tq = target_tensor.tensor_type.name
except Exception:
_tq = "?"
try:
_vq = vqtype if vraw is not None else vref.dtype
except Exception:
_vq = "?"
if ctx.VERBOSE:
_note(f"Voice '{vref.name}' is {_vq} {tuple(vref.shape)} → target '{tname}' is {_tq} {logical}")
if vraw is not None:
# GGUF quant voice: keep obtained quant (no upscale, no extra cycle).
try:
_vqtype = Q[vqtype]
except Exception:
_fail(f" ✗ Unknown voice quant '{vqtype}'.")
if tuple(vref.shape) != logical:
_fail(f" ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {logical}.")
if _quant_override in ("keep",):
# Transplant quant type as-is (may differ from target type — intentional, preserves quality).
if _vqtype.name != target_tensor.tensor_type.name and ctx.VERBOSE:
_note(f"Keeping voice quant {_vqtype.name} (target was {target_tensor.tensor_type.name}, not upscaling).")
_step(f"Casting onto {os.path.basename(target)} ({_vqtype.name}, kept)…")
from gguf.quants import quant_shape_to_byte_shape as _q2b_keep
_keep_shape = _q2b_keep(tuple(int(x) for x in vref.shape), _vqtype)
_need = 1
for _d in _keep_shape:
_need *= _d
_raw = np.frombuffer(vraw, dtype=np.uint8)
if _raw.size < _need:
_fail(f" ✗ Short voice data for transplant ({_raw.size}/{_need} bytes).")
q_head, out_qtype = _raw[:_need].reshape(_keep_shape), _vqtype
elif _quant_override == "target":
# Explicit: dequant once then quant to target (extra cycle, warn).
_warn(f"Re-quanting voice {_vqtype.name} → target {target_tensor.tensor_type.name} (extra dequant→quant cycle).")
from gguf.quants import dequantize as _deq2
from gguf.quants import quant_shape_to_byte_shape as _q2b
# vref.shape is logical already checked
_rshape = tuple(vref.shape)
try:
_f32v = _deq2(np.frombuffer(vraw, dtype=np.uint8).reshape(_q2b(_rshape, _vqtype)), _vqtype)
except Exception as e:
_fail(f" ✗ Could not dequant voice for --quant target: {e}")
if _is_delta:
_f32v = _fuse_delta(target_tensor, _f32v, strict=True)
t_quant0 = time.time()
_tq2 = target_tensor.tensor_type
if _tq2 in (Q.F32,):
q_head, out_qtype = _f32v.astype(np.float32), Q.F32
elif _tq2 in (Q.F16,):
q_head, out_qtype = _f32v.astype(np.float16), Q.F16
else:
try:
_bf = getattr(Q, "BF16", None)
except Exception:
_bf = None
if _bf is not None and _tq2 == _bf:
# BF16 target: gguf lib has no BF16 encoder here; keep F32 to avoid loss
q_head, out_qtype = _f32v.astype(np.float32), Q.F32
_warn(" BF16 target: keeping F32 to avoid loss (use --quant f16/bf16 explicitly if needed).")
else:
try:
q_head, out_qtype = quantize(_f32v, _tq2), _tq2
except Exception as e:
_fail(f" ✗ Could not quant to target {_tq2.name}: {e}")
_note(f"Re-quantized {_f32v.nbytes/1e6:.1f} MB → {_tq2.name} @ {(_f32v.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
else:
# Explicit q8_0/f16/f32/bf16 for quant voice: dequant once then quant.
from gguf.quants import dequantize as _deq2
from gguf.quants import quant_shape_to_byte_shape as _q2b
_rshape = tuple(vref.shape)
try:
_f32v = _deq2(np.frombuffer(vraw, dtype=np.uint8).reshape(_q2b(_rshape, _vqtype)), _vqtype)
except Exception as e:
_fail(f" ✗ Could not dequant voice for --quant {_quant_override}: {e}")
if _is_delta:
_f32v = _fuse_delta(target_tensor, _f32v, strict=True)
_qn = _quant_from_key(_quant_override)
if _qn is None:
_fail(f" ✗ Invalid --quant '{_quant_override}'.")
t_quant0 = time.time()
try:
q_head, out_qtype = _quantize_to_Q(_f32v, _qn)
except NotImplementedError as e:
_fail(f" ✗ Quant {getattr(_qn, 'name', _quant_override)} not implemented by gguf lib ({e}). Keeping original is recommended.")
except Exception as e:
_fail(f" ✗ Could not quant to {_quant_override}: {e}")
_note(f"Quantized {_f32v.nbytes/1e6:.1f} MB → {out_qtype.name} @ {(_f32v.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
else:
oriented = _orient(vf32, vref.shape, logical)
if oriented is None:
_fail(f" ✗ Voice shape {vref.shape} doesn't fit target tensor {tname} {logical}.")
t_fuse0 = time.time()
# Decide out quant: Q8_0 once only for safetensors high-fidelity heads; else keep target.
# vref.dtype here is F32/F16/BF16 (high-fidelity, since vraw is None).
_target_q = target_tensor.tensor_type
# Explicit override wins (any quant trivially supported via _quantize_to_Q).
_req = None
if _quant_override != "keep":
if _quant_override == "target":
_req = _target_q
else:
_req = _quant_from_key(_quant_override)
if _req is None:
_fail(f" ✗ Invalid --quant '{_quant_override}'.")
t_fuse0 = time.time()
if _is_delta:
# Fuse first (F32 math preserves quality), decide quant after.
fused = _fuse_delta(target_tensor, oriented, strict=True)
if _req is not None:
# Explicit --quant for delta (all quants trivial via helper).
_step(f"Fusing style into {tname} (delta + head → {_req.name})…")
t_quant0 = time.time()
try:
q_head, out_qtype = _quantize_to_Q(fused, _req)
except NotImplementedError as e:
_fail(f" ✗ Quant {_req.name} not implemented by gguf lib ({e}). Keeping original is recommended.")
except Exception as e:
_fail(f" ✗ Could not quant delta to {_req.name}: {e}")
_note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_req.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
elif _is_head:
# Head delta from high-fidelity: Q8_0 once (old behavior, with %32 fallback).
if fused.shape[-1] % 32 != 0:
_warn(f"Head last dim {fused.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
_step(f"Fusing style into {tname} (delta + head → F16 fallback)…")
q_head, out_qtype = fused.astype(np.float16), Q.F16
else:
_step(f"Fusing style into {tname} (delta + head → Q8_0)…")
t_quant0 = time.time()
try:
q_head, out_qtype = quantize(fused, Q.Q8_0), Q.Q8_0
except Exception as e:
_fail(f" ✗ Could not quant delta to Q8_0: {e}")
_note(f"Quantized {fused.nbytes/1e6:.1f} MB → Q8_0 @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
else:
# Generic delta: keep target quant (never force Q8_0). Never upscale implicitly.
_step(f"Fusing style into {tname} (delta + head → {_target_q.name}, kept)…")
t_quant0 = time.time()
if _target_q in (Q.F32,):
q_head, out_qtype = fused.astype(np.float32), Q.F32
elif _target_q in (Q.F16,):
q_head, out_qtype = fused.astype(np.float16), Q.F16
else:
try:
_b = getattr(Q, "BF16", None)
except Exception:
_b = None
if _b is not None and _target_q == _b:
q_head, out_qtype = fused.astype(np.float32), Q.F32
_warn(" BF16 target: keeping F32 to avoid loss (use --quant bf16 explicitly if needed).")
out_qtype = Q.F32
else:
try:
q_head, out_qtype = quantize(fused, _target_q), _target_q
except Exception as e:
_near = _nearest_encodable(_target_q)
if _near is not None:
_warn(f" {_target_q.name} not encodable ({_qerr(e)}); keeping {_near.name} (nearest size) instead of F16.")
q_head, out_qtype = _quantize_to_Q(fused, _near)
else:
_warn(f" Could not quant to {_target_q.name} ({_qerr(e)}), keeping F16.")
q_head, out_qtype = fused.astype(np.float16), Q.F16
try:
_note(f"Fused {fused.nbytes/1e6:.1f} MB → {out_qtype.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
except Exception:
pass
else:
# Not delta: voice replace (all quants trivial).
if _req is not None:
_step(f"Casting onto {os.path.basename(target)} ({_req.name})…")
fused = oriented
t_quant0 = time.time()
try:
q_head, out_qtype = _quantize_to_Q(fused, _req)
except NotImplementedError as e:
_fail(f" ✗ Quant {_req.name} not implemented by gguf lib ({e}).")
except Exception as e:
_fail(f" ✗ Could not quant to {_req.name}: {e}")
_note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_req.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
elif _is_head:
# lm_head high-fidelity: quant once to Q8_0 (preserve quality, Q4 saves little).
if oriented.shape[-1] % 32 != 0:
_warn(f"Head last dim {oriented.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
_step(f"Casting onto {os.path.basename(target)} (F16 fallback)…")
fused = oriented
q_head, out_qtype = fused.astype(np.float16), Q.F16
else:
_step(f"Casting onto {os.path.basename(target)} (Q8_0)…")
fused = oriented
t_quant0 = time.time()
try:
q_head, out_qtype = quantize(fused, Q.Q8_0), Q.Q8_0
except Exception as e:
sys.stdout.write("\n")
if ctx.VERBOSE:
_say(f" Details: {e}")
_fail(" ✗ Could not prepare the voiced model. Try again.")
_note(f"Quantized {fused.nbytes/1e6:.1f} MB → Q8_0 @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
else:
# Generic tensor: keep target quant (fix crash: F32 norms stay F32, never forced Q8_0).
_tq = _target_q
try:
_tqn = _tq.name
except Exception:
_tqn = str(_tq)
try:
_vqn = vref.dtype
except Exception:
_vqn = "?"
# Mismatch prompt only for safetensors non-head high-fidelity differing from target.
_need_prompt = (_vqn.upper() in ("F32", "F16", "BF16", "FP32", "FP16") and _tqn.upper() != _vqn.upper())
_custom_done = False
if _need_prompt and _quant_override == "keep" and not _noninteractive(args):
try:
_say(f" Target '{tname}' is {_tqn}, fetched is {_vqn}.")
_ans = input(f" Press enter to quantize to {_tqn} or enter custom quant (any like Q8_0/Q4_0/Q6_K/F16/F32/BF16): ").strip()
except (EOFError, KeyboardInterrupt):
_say(" ⚠ Cancelled.")
sys.exit(1)
_ans_n = _ans.strip().upper().replace("-", "_")
if _ans.strip() == "" or _ans_n == _tqn.upper():
pass
else:
_ck = _parse_quant_override(_ans)
_cq = _quant_from_key(_ck) if _ck not in (None, "keep", "target") else None
if _ck is None or (_cq is None and _ck not in ("keep", "target")):
_warn(f" Unknown quant '{_ans}', using target {_tqn}.")
elif _ck == "target" or _ans_n == _tqn.upper():
pass
else:
_warn(f"Forcing {_ans} for generic tensor (extra quant cycle).")
fused = oriented
t_quant0 = time.time()
try:
q_head, out_qtype = _quantize_to_Q(fused, _cq)
except NotImplementedError as e:
_fail(f" ✗ Quant {_cq.name} not implemented ({e}).")
except Exception as e:
_fail(f" ✗ Could not quant to {_cq.name}: {e}")
_note(f"Quantized {fused.nbytes/1e6:.1f} MB → {_cq.name} @ {(fused.nbytes/1e6)/max(time.time()-t_quant0,1e-6):.0f} MB/s")
_fd = time.time() - t_fuse0
if _fd >= 0.5:
_note(f"Fusing done in {_fd:.1f}s")
_custom_done = True
# keep-target path (default, silent when match, Enter=target when mismatch)
if not _custom_done:
_step(f"Casting '{vref.name}' ({_vqn}) onto '{tname}' ({_tqn}, kept)…")
fused = oriented
t_quant0 = time.time()
if _tq in (Q.F32,):
q_head, out_qtype = fused.astype(np.float32), Q.F32
elif _tq in (Q.F16,):
q_head, out_qtype = fused.astype(np.float16), Q.F16
else:
try:
_b = getattr(Q, "BF16", None)
except Exception:
_b = None
if _b is not None and _tq == _b:
q_head, out_qtype = fused.astype(np.float32), Q.F32
_warn(" BF16 target: keeping F32 to avoid loss (use --quant bf16 explicitly if needed).")
out_qtype = Q.F32
else:
try:
q_head, out_qtype = quantize(fused, _tq), _tq
except Exception as e:
_near = _nearest_encodable(_tq)
if _near is not None:
_warn(f" {_tqn} not encodable ({_qerr(e)}); keeping {_near.name} (nearest size) instead of F16.")
q_head, out_qtype = _quantize_to_Q(fused, _near)
else:
q_head, out_qtype = fused.astype(np.float16), Q.F16
try:
_mbk = fused.nbytes / 1e6
_note(f"Kept '{tname}': {_mbk:.1f} MB → {out_qtype.name}" + (f" @ {_mbk / max(time.time() - t_quant0, 1e-6):.0f} MB/s" if fused.nbytes >= 100000 else ""))
except Exception:
pass
try:
t_fuse0
except Exception:
t_fuse0 = time.time()
_fd = time.time() - t_fuse0
if _fd >= 0.5:
_note(f"Fusing done in {_fd:.1f}s")
return q_head, out_qtype
def _gguf_write_repl(target, reader, arch, repl, out_path, args, op, detail=None, _extra_names=()):
"""Single-pass GGUF write. repl maps tensor name -> (array, ggml_type).
detail (single-tensor callers): (tname, logical, vref, vraw, vqtype, meta) for rich messages."""
from gguf import GGUFWriter
from gguf.gguf_writer import GGUFValue
from gguf.constants import GGMLQuantizationType as Q
if not arch:
_warn(" Target architecture unknown — labeling output 'llama' (GGUF requires an arch).")
writer = GGUFWriter(out_path, arch or "llama", use_temp_file=True)
for field in reader.fields.values():
if field.name in ("general.architecture",) or field.name.startswith("GGUF."):
continue
try:
val = field.contents()
except Exception:
continue
vtype = field.types[0] if field.types else None
if vtype is None:
continue
subtype = field.types[-1] if len(field.types) > 1 else None
try:
writer.kv_data[0][field.name] = GGUFValue(value=val, type=vtype, sub_type=subtype)
except Exception:
try:
writer.add_key_value(field.name, val, vtype, subtype)
except Exception:
if ctx.VERBOSE:
_warn(f"Could not copy field {field.name}")
for t in reader.tensors:
if t.name in repl:
_qh, _qt = repl[t.name]
writer.add_tensor(t.name, _qh, raw_dtype=_qt)
elif t.tensor_type in (Q.F32, Q.F16):
writer.add_tensor(t.name, t.data)
else:
writer.add_tensor(t.name, t.data, raw_dtype=t.tensor_type)
for name in _extra_names:
if name in repl and name not in [t.name for t in reader.tensors]:
_qh, _qt = repl[name]
writer.add_tensor(name, _qh, raw_dtype=_qt)
# writing GGUF is the long wait (often 5-30s for 26GB) — show bar with speed, internet or local
_step(f"Writing {out_path} …")
# estimate total bytes for progress — use data.nbytes
try:
total_write = 8192
for t in reader.tensors:
if t.name in repl:
continue
try:
total_write += int(t.data.nbytes)
except Exception as e:
if ctx.VERBOSE:
_warn(f"progress estimate for {t.name} failed: {e}")
try:
for _qh, _qt in repl.values():
total_write += int(_qh.nbytes)
except Exception:
try:
for _qh, _qt in repl.values():
total_write += int(_qh.size * _qh.itemsize)
except Exception:
pass
if ctx.VERBOSE:
_note(f"Estimated write {total_write/1e6:.1f} MB ({len(repl)} replaced + {len([t for t in reader.tensors if t.name not in repl])} copied)")
except Exception as e:
if ctx.VERBOSE:
_warn(f"progress estimate failed: {e}")
total_write = 0
# ensure total_write is at least header + heads for small models, never 0 for progress
if total_write < 1000:
try:
total_write = 8192
for _qh, _qt in repl.values():
try:
total_write += int(_qh.nbytes)
except Exception:
pass
for t in reader.tensors:
if t.name not in repl:
try:
total_write += int(t.data.nbytes)
except Exception:
pass
except Exception:
pass
if total_write < 1000:
total_write = 0
t_write0 = time.time()
last_w = [0.0]
# progress poll for GGUF write (local disk, but user is waiting)
import threading
_stop_poll = [False]
_tiny_w = bool(total_write) and total_write < 100000
def _poll_gguf():
# poll output file (or temp file) size
target_path = getattr(writer, 'fname', out_path)
# GGUFWriter with use_temp_file=True writes to temp file with suffix
# fallback to out_path
poll_path = target_path if os.path.exists(target_path) else out_path
# try to find temp file if not yet at final path
while not _stop_poll[0]:
try:
cur = 0
if os.path.exists(poll_path):
cur = os.path.getsize(poll_path)
else:
# try temp file pattern
import glob
tmps = glob.glob(str(out_path) + "*")
for p in tmps:
if os.path.isfile(p):
try:
cur = max(cur, os.path.getsize(p))
except Exception:
pass
elapsed = time.time() - t_write0 or 1e-6
mbps = cur / elapsed / 1e6
pct = (cur / total_write * 100) if total_write else 0
if pct > 100:
pct = 100
if total_write:
msg = f" → writing {pct:3.0f}% · {cur/1e6:.1f}/{total_write/1e6:.1f} MB · {mbps:.0f} MB/s "
else:
msg = f" → writing {cur/1e6:.1f} MB · {mbps:.0f} MB/s "
sys.stdout.write(_transform(msg) + ("\n" if _progress_pref() else "\r"))
sys.stdout.flush()
except Exception:
pass
time.sleep(0.25)
_thr = threading.Thread(target=_poll_gguf, daemon=True)
if not _tiny_w:
_thr.start()
try:
writer.write_header_to_file()
writer.write_kv_data_to_file()
writer.write_tensors_to_file()
writer.close()
except Exception as e:
_stop_poll[0] = True
if not _tiny_w:
try:
_thr.join(timeout=0.5)
except Exception:
pass
sys.stdout.write("\n")
if ctx.VERBOSE:
_say(f" Details: {e}")
_fail(f" ✗ Could not save the voiced model to {out_path}.")
_stop_poll[0] = True
if not _tiny_w:
try:
_thr.join(timeout=0.5)
except Exception:
pass
sys.stdout.write("\n")
elapsed_w = time.time() - t_write0 or 1e-6
try:
final_sz = os.path.getsize(out_path)
_rate_w = f" @ {final_sz / elapsed_w / 1e6:.0f} MB/s" if final_sz >= 100000 else ""
_note(f"Wrote {final_sz / 1e6:.1f} MB{_rate_w}")
except Exception:
pass
file_size = os.path.getsize(out_path)
if detail is None:
summed = [(n, _qt.name if hasattr(_qt, "name") else str(_qt)) for n, (_qh, _qt) in repl.items()]
log_operation({"op": op, "target": target, "format": "gguf", "tensors": summed,
"output": out_path, "status": "ok"})
_ok(f"Done → {out_path}")
for n, _qt in summed:
_say(f" Cast '{n}' → {_qt} ({file_size/1e6:.1f} MB total).")
_say(f" Load it: llama serve -m {out_path}")
return
tname, logical, vref, vraw, vqtype, meta = detail
_qh0, out_qtype = repl[tname]
log_operation({"op": op, "target": target, "format": "gguf", "tensor": tname,
"dtype": out_qtype.name, "shape": list(logical), "output": out_path, "status": "ok"})
_ok(f"Done → {out_path}")
is_fallback = 'fallback_to_f16' in locals() and locals().get('fallback_to_f16')
_is_head_msg = _is_head_name(tname) or _is_head_name(getattr(vref, "name", None))
_label = "Head" if _is_head_msg else "Tensor"
if (meta or {}).get("type") == "delta" and is_fallback:
_say(f" Style fused and head kept as F16 (Q8_0 needs %32, fallback) ({file_size/1e6:.1f} MB).")
elif (meta or {}).get("type") == "delta":
# Delta head keeps Q8_0 for heads, keeps target for generic.
if _is_head_msg and out_qtype == Q.Q8_0:
_say(f" Style fused and head quantized to Q8_0 ({file_size/1e6:.1f} MB).")
else:
_say(f" Style fused into '{tname}' → {out_qtype.name}, kept ({file_size/1e6:.1f} MB).")
elif is_fallback:
_say(f" Head kept as F16 (Q8_0 needs last dim %32, fallback) ({file_size/1e6:.1f} MB).")
elif out_qtype == Q.Q8_0 and _is_head_msg:
_qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
_kept_q8 = (vraw is not None and isinstance(vqtype, str) and vqtype == Q.Q8_0.name and _qo in ("keep",))
if _kept_q8:
_say(f" Head kept as Q8_0 (from {vqtype}, {file_size/1e6:.1f} MB).")
else:
_say(f" Head quantized to Q8_0 ({file_size/1e6:.1f} MB).")
else:
_from = vqtype if vraw is not None else getattr(vref, "dtype", "?")
_say(f" {_label} '{tname}' kept as {out_qtype.name} (from {_from}, {file_size/1e6:.1f} MB).")
_say(f" Load it: llama serve -m {out_path}")
def _read_gguf_arch(reader):
"""Single arch-reader for GGUF targets. Returns the general.architecture string or None."""
import numpy as np
for field in reader.fields.values():
if field.name == "general.architecture":
try:
raw = field.parts[field.data[0]]
return raw.tobytes().decode() if isinstance(raw, np.ndarray) else str(raw)
except Exception:
return None
return None
def _confirm_arch_mismatch(varch, arch, args):
"""Single mismatch warn (+ confirm) shared by single and pack GGUF casts."""
if varch and arch and varch.lower() != arch.lower():
_warn(f"This voice was trained for a {varch} model, but the target is {arch}. It may not sound right.")
if not _confirm(" Continue anyway? [y/N]: ", args):
_note("Cancelled.")
sys.exit(1)
def _cast_gguf_speak(target, reader, arch, out_path, vf32, vref, meta, args, op, vraw=None, vqtype=None):
"""--speak: voice onto output.weight, invented if absent. GGUF only."""
import numpy as np
from gguf.constants import GGMLQuantizationType as Q
from gguf.quants import quantize, quant_shape_to_byte_shape
if (meta or {}).get("type") == "delta":
_fail(" ✗ --speak cannot place a delta voice (differences are not weights).")
names = [t.name for t in reader.tensors]
ot = next((t for t in reader.tensors if t.name == "output.weight"), None)
et = next((t for t in reader.tensors if t.name == "token_embd.weight"), None)
if ot is not None:
olog = tuple(reversed(tuple(int(x) for x in ot.shape)))
invented = False
elif et is not None:
olog = tuple(reversed(tuple(int(x) for x in et.shape)))
invented = True
else:
_fail(" ✗ Target has no output.weight or token_embd.weight to mirror geometry from.")
varch = (meta or {}).get("architecture")
_confirm_arch_mismatch(varch, arch, args)
tname = "output.weight"
if vraw is not None:
_vqtype = Q[vqtype]
if tuple(vref.shape) != olog:
_fail(f" ✗ Voice shape {vref.shape} doesn't fit {tname} {olog}.")
_qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
if _qo not in ("keep",):
fitted = _gguf_raw_to_f32(vraw, _vqtype, tuple(reversed(tuple(int(x) for x in vref.shape))))
q_head, out_qtype = _quantize_to_Q(np.ascontiguousarray(fitted), Q[_qo.upper()])
else:
_need = 1
for _d in quant_shape_to_byte_shape(tuple(int(x) for x in vref.shape), _vqtype):
_need *= _d
_raw = np.frombuffer(vraw, dtype=np.uint8)
if _raw.size < _need:
_fail(f" ✗ Voice data too short for {vqtype} {tuple(vref.shape)}.")
q_head = np.ascontiguousarray(_raw[:_need].reshape(
quant_shape_to_byte_shape(tuple(int(x) for x in vref.shape), _vqtype)))
out_qtype = _vqtype
else:
fitted = _orient(vf32, vref.shape, olog)
if fitted is None:
_fail(f" ✗ Voice shape {vref.shape} doesn't fit {tname} {olog}.")
fitted = np.ascontiguousarray(fitted)
_qo = _parse_quant_override(getattr(args, "quant", None) or "keep")
if _qo is None:
_fail(f" ✗ Invalid --quant '{getattr(args, 'quant', None)}'.")
if _qo not in ("keep",):
q_head, out_qtype = _quantize_to_Q(fitted, Q[_qo.upper()])
elif fitted.shape[-1] % 32 == 0:
q_head, out_qtype = quantize(fitted, Q.Q8_0), Q.Q8_0
else:
_note(f" Head last dim {fitted.shape[-1]} not ×32 — Q8_0 needs %32, keeping F16 (larger, compatible).")
q_head, out_qtype = fitted.astype(np.float16), Q.F16
if invented:
_note(" speak: voice → invented output.weight (file untied)")
else:
_note(" speak: voice → output.weight")
_gguf_write_repl(target, reader, arch, {tname: (q_head, out_qtype)}, out_path, args, op,
detail=(tname, olog, vref, vraw, vqtype, meta),
_extra_names=(tname,) if invented else ())
def _cast_gguf(target, out_path, vf32, vref, meta, args, op, vraw=None, vqtype=None, strict_target=False, target_name=None):
import numpy as np
from gguf import GGUFReader
reader = GGUFReader(target)
arch = _read_gguf_arch(reader)
if getattr(args, "speak", False):
_cast_gguf_speak(target, reader, arch, out_path, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
return
target_tensor = None
if target_name is not None:
# Explicitly routed (requested subset): honor it, never fall back.
for t in reader.tensors:
if t.name == target_name:
target_tensor = t
break
if target_tensor is None:
_fail(f" ✗ Resolved target '{target_name}' for '{vref.name}' not in {target}.")
if target_tensor is None and vref.name:
for t in reader.tensors:
if t.name == vref.name:
target_tensor = t
break
if target_tensor is None and strict_target:
_fail(f" ✗ Tensor '{vref.name}' not found in target {target}. Voicepack requires exact match (not mapping to another tensor).")
if target_tensor is None:
for preferred in ("output.weight", "token_embd.weight"):
for t in reader.tensors:
if t.name == preferred:
target_tensor = t
break
if target_tensor is not None:
break
if target_tensor is None:
_fail(" ✗ Target has no output.weight or token_embd.weight to receive the voice.")
tname = target_tensor.name
logical = tuple(reversed(tuple(int(x) for x in target_tensor.shape)))
# arch mismatch warning
varch = (meta or {}).get("architecture")
_confirm_arch_mismatch(varch, arch, args)
q_head, out_qtype = _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
_gguf_write_repl(target, reader, arch, {tname: (q_head, out_qtype)}, out_path, args, op,
detail=(tname, logical, vref, vraw, vqtype, meta))
# GGUF (llama.cpp) <-> safetensors (HF) tensor-name aliases, matched by
# (block index, role). Exact name match always wins; aliases only fire on a
# miss, and only when exactly one candidate matches. Anything else fails loud
# at the call site — never a silent mapping to another tensor.
_ROLE_ALIASES = (
("embed_tokens.weight", "token_embd.weight"),
("word_embeddings.weight", "token_embd.weight"),
("wte.weight", "token_embd.weight"),
("output_layer.weight", "output.weight"),
("lm_head.weight", "output.weight"),
("final_layernorm.weight", "output_norm.weight"),
("self_attention.dense.weight", "attn_output.weight"),
("mlp.dense_4h_to_h.weight", "ffn_down.weight"),
("model.norm.weight", "output_norm.weight"),
("ln_f.weight", "output_norm.weight"),
("input_layernorm.weight", "attn_norm.weight"),
("self_attn.q_proj.weight", "attn_q.weight"),
("self_attn.k_proj.weight", "attn_k.weight"),
("self_attn.v_proj.weight", "attn_v.weight"),
("self_attn.o_proj.weight", "attn_o.weight"),
("self_attn.o_proj.weight", "attn_output.weight"),
("post_attention_layernorm.weight", "post_attention_norm.weight"),
("mlp.down_proj.weight", "ffn_down.weight"),
("experts.down_proj", "ffn_down_exps.weight"),
("mlp.up_proj.weight", "ffn_up.weight"),
)
def _block_index(name):
import re
m = re.search(r"(?<![A-Za-z0-9_])(?:layers|layer|blk|blks|blocks|h)\.(\d+)\b", name)
if m:
return int(m.group(1))
m = re.search(r"\.(\d+)\.", name)
return int(m.group(1)) if m else None
def _pack_aliases(vname, target_names):
"""All alias matches for vname in target_names (either naming direction).
Exact match is handled by callers first; this only fires on a miss."""
hits = []
for st_suf, gg_suf in _ROLE_ALIASES:
for t in target_names:
st_v = vname.endswith(st_suf)
gg_v = vname.endswith(gg_suf)
st_t = t.endswith(st_suf)
gg_t = t.endswith(gg_suf)
if (st_v and gg_t) or (gg_v and st_t):
if _block_index(vname) == _block_index(t) and t not in hits:
hits.append(t)
return hits
def _role_key(name):
"""(block_index, trailing_role) canonical key for same-format matching.
Trailing role is everything after the block number (e.g.
'self_attn.o_proj.weight'); unblocked names fall back to their last two
components (e.g. 'embed_tokens.weight'). The .weight/.bias suffix stays
in the key so a bias never matches a weight. The block keyword
(layers/blk/...) is left-guarded against mid-word hits; the generic
dotted-number fallback still fires on version-like segments, so the
elem-count gate remains the real false-friend veto."""
import re
m = re.search(r"(?<![A-Za-z0-9_])(?:layers|layer|blk|blks|blocks|h)\.(\d+)\b", name)
if m:
return (int(m.group(1)), name[m.end():].strip("."))
m = re.search(r"\.(\d+)\.", name)
if m:
return (int(m.group(1)), name[m.end():].strip("."))
parts = name.split(".")
return (None, ".".join(parts[-2:]))
def _same_role(vname, target_names):
"""Targets sharing vname's (block, trailing role). Cross-format pairs
(ST suffix vs GGUF suffix) never collide here — _ROLE_ALIASES owns those.
Only fires on a miss; callers gate on elem-count and uniqueness."""
want = _role_key(vname)
return [t for t in target_names if _role_key(t) == want]
def _cast_gguf_pack(target, out_path, items, args, op):
"""Single-pass pack cast onto GGUF. items: [(vf32, vref, meta, vraw, vqtype)]."""
import numpy as np
from gguf import GGUFReader
reader = GGUFReader(target)
arch = _read_gguf_arch(reader)
varch = (items[0][2] or {}).get("architecture") if items else None
_confirm_arch_mismatch(varch, arch, args)
_names = [t.name for t in reader.tensors]
_counts = {}
for t in reader.tensors:
try:
n = 1
for d in t.shape:
n *= int(d)
_counts[t.name] = n
except Exception:
pass
_map = _resolve_pack_targets([(ref.name, ref.shape) for (_, ref, _, _, _) in items],
_names, target, _counts)
repl = {}
for vf32, vref, meta, vraw, vqtype in items:
target_tensor = next(t for t in reader.tensors if t.name == _map[vref.name])
tname = target_tensor.name
logical = tuple(reversed(tuple(int(x) for x in target_tensor.shape)))
q_head, out_qtype = _gguf_fuse_one(target, target_tensor, tname, logical, vf32, vref, meta, args, op, vraw=vraw, vqtype=vqtype)
repl[tname] = (q_head, out_qtype)
_gguf_write_repl(target, reader, arch, repl, out_path, args, op)
def _resolve_pack_targets(pairs, target_names, target, counts=None):
"""Map every voice tensor to a target tensor (exact, else one alias,
else one same-role same-format match).
Resolves ALL up front so a pack fails once with every miss — never
fix-forward one at a time. Same-size targets are offered as hints only.
pairs: [(voice_name, voice_shape)]; counts: {target_name: elem_count}.
Voice shapes are logical-as-is (GgufSource.ref already reverses);
only elem-counts are compared, so transposes pass and nothing double-flips."""
def _want(vshape):
if vshape is None:
return None
n = 1
for d in vshape:
n *= int(d)
return n
mapping, misses = {}, []
for vname, vshape in pairs:
if vname in target_names:
mapping[vname] = vname
continue
cands = _pack_aliases(vname, target_names)
if len(cands) == 1:
mapping[vname] = cands[0]
_note(f" Aliased '{vname}' → '{cands[0]}' ({target} naming).")
elif len(cands) > 1:
misses.append(f" '{vname}' is ambiguous: {', '.join(cands)}")
else:
roles = [t for t in _same_role(vname, target_names)
if _want(vshape) is None or counts is None
or counts.get(t, _want(vshape)) == _want(vshape)]
if len(roles) == 1:
mapping[vname] = roles[0]
_note(f" Aliased '{vname}' → '{roles[0]}' (same role).")
continue
if len(roles) > 1:
misses.append(f" '{vname}' is ambiguous: {', '.join(roles)}")
continue
hint = ""
if counts is not None and vshape is not None:
same = [t for t in target_names if counts.get(t) == _want(vshape)][:3]
if same:
hint = f" (same size: {', '.join(same)})"
misses.append(f" '{vname}' not found{hint}")
if misses:
_fail(f" ✗ {len(misses)} tensor(s) have no match in target {target}:\n"
+ "\n".join(misses)
+ "\n Voicepack requires exact match (not mapping to another tensor).")
return mapping
def _cast_pack(target, src_path, out_path, args, op, excludes=(), meta=None, order=()):
"""Cast every tensor in a pack file onto target (minus excludes).
GGUF and safetensors targets each get one write pass (repl-dict fused first).
Delta marker on the pack is detected once and threaded into every tensor.
order (requested subset): process in the user's order, not file order."""
if getattr(args, "speak", False) and not _is_gguf_file(target):
_fail(" ✗ --speak needs a .gguf target. Re-run without --speak for .safetensors, or point cast at a .gguf file.")
src = open_source(str(src_path))
all_names = [n for n in src.names() if n != "delta.voice.marker"]
for e in excludes:
if e not in all_names and not any(e in n for n in all_names):
_warn(f" --exclude '{e}' not in pack (has {len(all_names)} tensors). Ignoring.")
resolved_ex = set()
for e in excludes:
if e in all_names:
resolved_ex.add(e)
else:
_m = [n for n in all_names if e in n]
if len(_m) == 1:
resolved_ex.add(_m[0])
elif len(_m) > 1:
_fail(f" ✗ --exclude '{e}' is ambiguous: {', '.join(_m)}. Name one tensor exactly.")
to_cast = [n for n in all_names if n not in resolved_ex]
if order:
_pos = {n: i for i, n in enumerate(order)}
to_cast.sort(key=lambda n: (_pos.get(n, len(_pos)), n))
if not to_cast:
_fail(" ✗ --exclude removes all tensors. Nothing to cast.")
_step(f"Casting {len(to_cast)} tensors from {Path(src_path).name} onto {Path(target).name}…")
if resolved_ex:
_note(f"Excluding {len(resolved_ex)}: {', '.join(sorted(resolved_ex))}")
is_delta_file = str(src_path).endswith(".safetensors") and _is_delta_safetensors(str(src_path))
if is_delta_file:
_note("Detected delta voice — fusing style (delta + head)…")
if meta is None and is_delta_file:
meta = {"type": "delta", "source": str(src_path)}
items = []
for i, tname in enumerate(to_cast):
f32, ref, raw, qtype = _load_tensor(src_path, tname, _src=src)
_step(f"[{i+1}/{len(to_cast)}] {tname} ({ref.dtype} {tuple(ref.shape)})…")
items.append((f32, ref, meta, raw, qtype))
if _is_gguf_file(target):
_cast_gguf_pack(target, out_path, items, args, op)
return
_header, _tensors = read_safetensors(target)
_st_names = list(_tensors.keys())
_st_counts = {}
for _n, _info in _tensors.items():
try:
_c = 1
for _d in _info["shape"]:
_c *= int(_d)
_st_counts[_n] = _c
except Exception:
pass
_st_map = _resolve_pack_targets([(ref.name, ref.shape) for (_, ref, _, _, _) in items],
_st_names, target, _st_counts)
repl = {}
for f32, ref, m, raw, qtype in items:
tname, data, tdtype, tshape = _cast_st_prepare(target, _tensors, f32, ref, m, args,
vraw=raw, vqtype=qtype, strict_target=True,
target_name=_st_map[ref.name])
repl[tname] = (data, tdtype, tshape)
_step(f"Replacing {len(repl)} tensors in {Path(target).name}…")
_safetensors_write_repl(target, repl, out_path)
file_size = os.path.getsize(out_path)
log_operation({"op": op, "target": target, "format": "safetensors",
"tensors": [(n, d) for n, (_b, d, _s) in repl.items()],
"output": out_path, "status": "ok"})
_ok(f"Done → {out_path}")
for n, (_b, d, _s) in repl.items():
_say(f" Tensor '{n}' replaced at {d} ({file_size/1e6:.1f} MB total).")
def _parse_tensor_list(args):
# Tensor positionals (get/cast tensors...), with comma-split for shell convenience.
raw = getattr(args, "tensors", None) or []
if isinstance(raw, str):
raw = [raw]
# Split comma-separated entries, strip, drop empties.
out = []
for r in raw:
if r is None:
continue
for part in str(r).split(","):
part = part.strip()
if part:
out.append(part)
return out
def _parse_exclude_list(args):
raw = getattr(args, "excludes", None) or getattr(args, "exclude", None) or []
if isinstance(raw, str):
raw = [raw]
out = []
for r in (raw or []):
for part in str(r).split(","):
part = part.strip()
if part:
out.append(part)
return out
def _resolve_one_tensor(req, names):
# Exact match first, then single substring match, else fail. Returns resolved name.
if req in names:
return req
matches = [n for n in names if req and req in n]
if len(matches) == 1:
return matches[0]
if len(matches) > 1:
_fail(f" ✗ '{req}' is ambiguous: {', '.join(matches[:8])}…")
_fail(f" ✗ Tensor '{req}' not found. Try one of: {', '.join(names[:10])}")
def _tensor_src_meta(src_path):
"""File-level delta detection for the tensor path (the marker lives in the file; no voice.json)."""
if str(src_path).endswith(".safetensors"):
try:
if _is_delta_safetensors(str(src_path)):
return {"type": "delta", "source": str(src_path)}
except Exception:
pass
return None
def _load_tensor(path, name=None, _src=None):
src = _src if _src is not None else open_source(str(path))
names = src.names()
if not names:
_fail(f" ✗ No tensors found in {path}.")
if name is None:
name = "output.weight" if "output.weight" in names else names[0]
if name not in names:
matches = [n for n in names if name in n]
if len(matches) == 1:
name = matches[0]
else:
_fail(f" ✗ Tensor '{name}' not found in {path}.")
ref = src.ref(name)
if src.kind == "gguf" and _is_ggml_quant(ref.dtype):
return None, ref, src.read_raw(name), ref.dtype
return src.read_f32(name), ref, None, None
|