Private review: measured five-language release, Hindi sentence mode and custom GGUFs
a449848 verified Download python/gguf_read.py from sourojitdotcom/AksharaLP-MultiIndic-8M: direct link, hf CLI and curl.
- Browser
- Download file 16.4 kB
-
https://huggingface.co/sourojitdotcom/AksharaLP-MultiIndic-8M/resolve/main/python/gguf_read.py
- Command line
-
hf download hf://sourojitdotcom/AksharaLP-MultiIndic-8M/python/gguf_read.py
-
curl -L -o gguf_read.py https://huggingface.co/sourojitdotcom/AksharaLP-MultiIndic-8M/resolve/main/python/gguf_read.py
16.4 kB
| """Minimal, dependency-light reader for the GGUF files shipped with AksharaLP-MultiIndic-8M. | |
| GGUF here is used as what it is: a portable, single-file tensor container -- a header, typed | |
| key/value metadata, a tensor table, and aligned little-endian tensor data. It is NOT a | |
| llama.cpp-loadable model: AksharaLP is a custom character-level encoder-decoder transformer with a | |
| bespoke decoding stack (lexicon-trie candidate beam, hand-built candidate features, gates, MLP | |
| reranker), and llama.cpp has no such architecture. This reader is the decoder-side tool: it returns | |
| the weights our reference implementation (python/bhc/xlit_*.py) expects. | |
| Only numpy is required. No torch, no llama.cpp, no gguf/gguf-py. | |
| python gguf_read.py aksharalp-1m-v2-f16.gguf # summary | |
| python gguf_read.py --json aksharalp-1m-v2-q4.gguf # metadata + tensor table | |
| import gguf_read | |
| w = gguf_read.load("aksharalp-1m-v2-f16.gguf") # -> Weights(cfg, state_dict, metadata) | |
| w.state_dict["semb.weight"].shape # (30, 128) float32 | |
| Two container layouts are understood, both GGUF v3: | |
| * aksharalp-1m-v2-q4.gguf one I8 tensor, ``aksharalp.q4.blob``: the byte-exact xlit.q4 file | |
| (4-byte header length, JSON tensor table, then the payload). Decoded | |
| here exactly as bhc/xlit_q.py:load_q does. | |
| * aksharalp-1m-v2-f16.gguf every weight under its state_dict name at F16. | |
| """ | |
| import json | |
| import struct | |
| import sys | |
| import numpy as np | |
| MAGIC = b"GGUF" | |
| SUPPORTED_VERSIONS = (3,) | |
| # GGUF metadata value types | |
| GGUF_UINT8, GGUF_INT8, GGUF_UINT16, GGUF_INT16 = 0, 1, 2, 3 | |
| GGUF_UINT32, GGUF_INT32, GGUF_FLOAT32, GGUF_BOOL = 4, 5, 6, 7 | |
| GGUF_STRING, GGUF_ARRAY, GGUF_UINT64, GGUF_INT64, GGUF_FLOAT64 = 8, 9, 10, 11, 12 | |
| # GGML tensor types we can read: F32, F16, I8 (the three this model uses) | |
| _GGML_F32, _GGML_F16, _GGML_I8 = 0, 1, 24 | |
| _TENSOR_DTYPES = {_GGML_F32: np.dtype("<f4"), _GGML_F16: np.dtype("<f2"), _GGML_I8: np.dtype("i1")} | |
| _TENSOR_NAMES = {_GGML_F32: "F32", _GGML_F16: "F16", _GGML_I8: "I8"} | |
| _KV_FMT = {GGUF_UINT8: "<B", GGUF_INT8: "<b", GGUF_UINT16: "<H", GGUF_INT16: "<h", | |
| GGUF_UINT32: "<I", GGUF_INT32: "<i", GGUF_FLOAT32: "<f", GGUF_BOOL: "?", | |
| GGUF_UINT64: "<Q", GGUF_INT64: "<q", GGUF_FLOAT64: "<d"} | |
| class GGUFError(ValueError): | |
| """Raised for anything that is not a well-formed GGUF v3 file we can read.""" | |
| class _Cursor: | |
| def __init__(self, buf, pos=0): | |
| self.buf, self.pos = buf, pos | |
| def take(self, n): | |
| if n < 0 or self.pos + n > len(self.buf): | |
| raise GGUFError("truncated file: wanted %d bytes at %d, file is %d" % (n, self.pos, len(self.buf))) | |
| b = self.buf[self.pos:self.pos + n] | |
| self.pos += n | |
| return b | |
| def unpack(self, fmt): | |
| n = struct.calcsize(fmt) | |
| return struct.unpack(fmt, self.take(n))[0] | |
| def string(self): | |
| """GGUF strings: u64 length + UTF-8 bytes (not NUL-terminated).""" | |
| n = self.unpack("Q") | |
| return self.take(n).decode("utf-8") | |
| class Tensor: | |
| """One tensor: name, logical shape (GGML order reversed back to row-major), dtype, raw bytes.""" | |
| __slots__ = ("name", "shape", "type_id", "offset", "data") | |
| def __init__(self, name, shape, type_id, offset, data): | |
| self.name, self.shape, self.type_id, self.offset, self.data = name, shape, type_id, offset, data | |
| def dtype(self): | |
| return _TENSOR_DTYPES[self.type_id] | |
| def type_name(self): | |
| return _TENSOR_NAMES[self.type_id] | |
| def array(self): | |
| return np.frombuffer(self.data, dtype=self.dtype).reshape(self.shape) | |
| def __repr__(self): | |
| return "<Tensor %s %s %s>" % (self.name, self.type_name, tuple(self.shape)) | |
| class GGUF: | |
| """Parsed GGUF file: header fields, typed metadata, tensor table, tensor data.""" | |
| def __init__(self, path): | |
| self.path = str(path) | |
| with open(self.path, "rb") as f: | |
| self.buf = f.read() | |
| self._parse() | |
| # ---------------------------------------------------------------- parsing | |
| def _parse(self): | |
| c = _Cursor(self.buf) | |
| self.magic = c.take(4) | |
| if self.magic != MAGIC: | |
| raise GGUFError("%s: not a GGUF file (magic %r)" % (self.path, self.magic)) | |
| self.version = c.unpack("I") | |
| if self.version not in SUPPORTED_VERSIONS: | |
| raise GGUFError("%s: GGUF version %d not supported (this reader knows %s)" | |
| % (self.path, self.version, SUPPORTED_VERSIONS)) | |
| self.tensor_count = c.unpack("Q") | |
| self.kv_count = c.unpack("Q") | |
| self.metadata = {} | |
| for _ in range(self.kv_count): | |
| key = c.string() | |
| self.metadata[key] = self._value(c) | |
| self.alignment = int(self.metadata.get("general.alignment", 32)) | |
| if self.alignment <= 0 or self.alignment & (self.alignment - 1): | |
| raise GGUFError("%s: general.alignment %d is not a power of two" % (self.path, self.alignment)) | |
| infos = [] | |
| for _ in range(self.tensor_count): | |
| name = c.string() | |
| n_dims = c.unpack("I") | |
| if n_dims > 8: | |
| raise GGUFError("%s: tensor %s declares %d dims" % (self.path, name, n_dims)) | |
| # GGML stores dims fastest-varying first; reverse to get the row-major shape. | |
| dims = [c.unpack("Q") for _ in range(n_dims)] | |
| type_id = c.unpack("I") | |
| offset = c.unpack("Q") | |
| infos.append((name, list(reversed(dims)), type_id, offset)) | |
| pad = (-c.pos) % self.alignment | |
| self.data_start = c.pos + pad | |
| if self.data_start > len(self.buf): | |
| raise GGUFError("%s: tensor info block overruns the file" % self.path) | |
| self.tensors = {} | |
| self.order = [] | |
| for name, shape, type_id, offset in infos: | |
| if type_id not in _TENSOR_DTYPES: | |
| raise GGUFError("%s: tensor %s has GGML type %d, which this reader does not decode " | |
| "(only F32/F16/I8 are used by these files)" % (self.path, name, type_id)) | |
| if name in self.tensors: | |
| raise GGUFError("%s: duplicate tensor name %r" % (self.path, name)) | |
| numel = int(np.prod(shape, dtype=np.int64)) if shape else 1 | |
| nbytes = numel * _TENSOR_DTYPES[type_id].itemsize | |
| start = self.data_start + offset | |
| end = start + nbytes | |
| if start < self.data_start or end > len(self.buf): | |
| raise GGUFError("%s: tensor %s spans [%d, %d) outside the data section [%d, %d)" | |
| % (self.path, name, start, end, self.data_start, len(self.buf))) | |
| if offset % self.alignment: | |
| raise GGUFError("%s: tensor %s offset %d is not aligned to %d" | |
| % (self.path, name, offset, self.alignment)) | |
| self.tensors[name] = Tensor(name, shape, type_id, offset, self.buf[start:end]) | |
| self.order.append(name) | |
| def _value(self, c, vtype=None): | |
| if vtype is None: | |
| vtype = c.unpack("I") | |
| if vtype == GGUF_STRING: | |
| return c.string() | |
| if vtype == GGUF_ARRAY: | |
| sub = c.unpack("I") | |
| n = c.unpack("Q") | |
| return [self._value(c, sub) for _ in range(n)] | |
| if vtype in _KV_FMT: | |
| return c.unpack(_KV_FMT[vtype]) | |
| raise GGUFError("%s: unknown metadata value type %d" % (self.path, vtype)) | |
| # ---------------------------------------------------------------- checks | |
| def validate(self): | |
| """Structural checks; returns a list of human-readable results, raises GGUFError on failure.""" | |
| notes = [] | |
| if len(self.tensors) != self.tensor_count: | |
| raise GGUFError("tensor count mismatch: header says %d, parsed %d" % (self.tensor_count, len(self.tensors))) | |
| notes.append("magic GGUF, version %d" % self.version) | |
| notes.append("%d tensors, %d metadata keys" % (self.tensor_count, len(self.metadata))) | |
| notes.append("alignment %d, data section at byte %d" % (self.alignment, self.data_start)) | |
| declared = self.metadata.get("general.architecture") | |
| if declared is None: | |
| raise GGUFError("general.architecture is missing") | |
| for name in self.order: | |
| t = self.tensors[name] | |
| if t.offset + len(t.data) > len(self.buf) - self.data_start: | |
| raise GGUFError("tensor %s overruns the data section" % name) | |
| notes.append("all tensor offsets aligned and inside the data section") | |
| arch = self.metadata.get("general.architecture") | |
| notes.append("general.architecture = %r (custom; not a llama.cpp architecture)" % arch) | |
| return notes | |
| # ---------------------------------------------------------------- weights | |
| def state_dict(self): | |
| """Return {state_dict name: float32 ndarray} for the reference XlitTransformer.""" | |
| cfg, sd = self._weights() | |
| return cfg, sd | |
| def _weights(self): | |
| if "aksharalp.nlm.blob" in self.tensors: | |
| return _weights_from_nlm_blob(bytes(self.tensors["aksharalp.nlm.blob"].data)) | |
| if _BLOB_TENSOR in self.tensors: | |
| return _weights_from_q4_blob(bytes(self.tensors[_BLOB_TENSOR].data)) | |
| cfg = _cfg_from_metadata(self.metadata) | |
| sd = {} | |
| for name in self.order: | |
| if name.startswith(RERANKER_PREFIX): | |
| continue # reranker matrices are carried separately; see reranker() | |
| a = self.tensors[name].array() | |
| sd[name] = np.ascontiguousarray(a, dtype=np.float32) | |
| return cfg, sd | |
| def reranker(self): | |
| """{('core'|'pack', 'w1'|'b1'|'w2'|'b2'): float32 ndarray} from the F16 variant, else {}.""" | |
| out = {} | |
| for name in self.order: | |
| if not name.startswith(RERANKER_PREFIX): | |
| continue | |
| _, part, param = name.split(".") | |
| out[(part, param)] = np.ascontiguousarray(self.tensors[name].array(), dtype=np.float32) | |
| return out | |
| _BLOB_TENSOR = "aksharalp.q4.blob" | |
| RERANKER_PREFIX = "reranker." | |
| def _cfg_from_metadata(md): | |
| if "aksharalp.config_json" in md: return json.loads(md["aksharalp.config_json"]) | |
| def g(key, default=None): | |
| v = md.get("aksharalp." + key, default) | |
| if v is None: | |
| raise GGUFError("metadata is missing aksharalp.%s" % key) | |
| return v | |
| cfg = {"d": int(g("d_model")), "h": int(g("n_heads")), "enc": int(g("encoder_layers")), | |
| "dec": int(g("decoder_layers")), "ff": int(g("ff_dim")), "p": float(g("dropout", 0.1))} | |
| return cfg | |
| def _weights_from_q4_blob(blob): | |
| """Decode the carried xlit.q4 bytes exactly as python/bhc/xlit_q.py:load_q does.""" | |
| n = int.from_bytes(blob[:4], "little") | |
| meta = json.loads(blob[4:4 + n]) | |
| data = blob[4 + n:] | |
| bits_default, G = int(meta["bits"]), int(meta["G"]) | |
| cfg = dict(meta["cfg"]) | |
| sd = {} | |
| for k, shape, kind, off, ln in meta["tensors"]: | |
| cnt = int(np.prod(shape)) | |
| if kind.startswith("q"): | |
| bits = int(kind[1:]) if len(kind) > 1 else bits_default | |
| if bits == 4: | |
| raw = np.frombuffer(data, np.uint8, ln, off) | |
| q = np.stack([raw >> 4, raw & 15], 1).reshape(-1)[:cnt].astype(np.int16) - 8 | |
| elif bits == 8: | |
| q = np.frombuffer(data, np.int8, ln, off).astype(np.int16) | |
| else: | |
| raise GGUFError("unsupported q width %d for %s" % (bits, k)) | |
| r, c = shape | |
| ng = (c + G - 1) // G | |
| sc = np.frombuffer(data, np.float16, r * ng, off + ln).astype(np.float32).reshape(r, ng) | |
| w = q.reshape(r, c).astype(np.float32) * np.repeat(sc, G, 1)[:, :c] | |
| sd[k] = w | |
| elif kind == "f16": | |
| sd[k] = np.frombuffer(data, np.float16, cnt, off).astype(np.float32).reshape(shape).copy() | |
| else: | |
| raise GGUFError("unsupported tensor kind %r for %s" % (kind, k)) | |
| return cfg, sd | |
| class Weights: | |
| """Weights + config, in the shape the reference implementation consumes. | |
| ``state_dict`` maps state_dict names (as in python/bhc/xlit_q.py:load_q) to float32 ndarrays; | |
| ``cfg`` is the XlitTransformer config. To build the torch model the reference uses: | |
| import torch | |
| from bhc.xlit_model import XlitTransformer | |
| m = XlitTransformer(**w.cfg) | |
| m.load_state_dict({k: torch.from_numpy(v) for k, v in w.state_dict.items()}) | |
| m.eval() | |
| """ | |
| def __init__(self, cfg, state_dict, metadata=None, source=None, reranker=None): | |
| self.cfg, self.state_dict, self.metadata, self.source = cfg, state_dict, metadata or {}, source | |
| self.reranker = reranker or {} | |
| def n_params(self): | |
| return int(sum(v.size for v in self.state_dict.values())) | |
| def __repr__(self): | |
| return "<Weights %s: %d tensors, %d params>" % (self.source, len(self.state_dict), self.n_params) | |
| def load(path): | |
| """Parse either GGUF variant and return Weights (cfg + state_dict of float32 arrays).""" | |
| g = GGUF(path) | |
| cfg, sd = g.state_dict() | |
| return Weights(cfg, sd, g.metadata, g.path, g.reranker()) | |
| def tensors(path): | |
| """Yield (name, shape, type_name, float32 array) for every tensor in the file.""" | |
| g = GGUF(path) | |
| for name in g.order: | |
| t = g.tensors[name] | |
| yield name, tuple(t.shape), t.type_name, t.array() | |
| def q4_blob(path): | |
| """Bytes of the carried xlit.q4 file from the compact GGUF (byte-identical to the original).""" | |
| g = GGUF(path) | |
| if _BLOB_TENSOR not in g.tensors: | |
| raise GGUFError("%s carries no %s tensor (that is the F16 variant)" % (path, _BLOB_TENSOR)) | |
| return bytes(g.tensors[_BLOB_TENSOR].data) | |
| def main(argv): | |
| args = [a for a in argv[1:] if not a.startswith("-")] | |
| flags = {a for a in argv[1:] if a.startswith("-")} | |
| if not args: | |
| print(__doc__) | |
| return 2 | |
| g = GGUF(args[0]) | |
| if "--json" in flags: | |
| print(json.dumps({"version": g.version, "tensors": g.tensor_count, | |
| "metadata": g.metadata, | |
| "tensor_table": [{"name": n, "shape": g.tensors[n].shape, | |
| "type": g.tensors[n].type_name, | |
| "offset": g.tensors[n].offset} for n in g.order]}, | |
| indent=1, ensure_ascii=False)) | |
| return 0 | |
| for line in g.validate(): | |
| print(line) | |
| print("general.name = %r" % g.metadata.get("general.name")) | |
| print("-- tensors --") | |
| for n in g.order: | |
| t = g.tensors[n] | |
| print(" %-28s %-4s %-18s %d B" % (n, t.type_name, tuple(t.shape), len(t.data))) | |
| w = load(args[0]) | |
| print("-- weights --") | |
| print("cfg = %s" % (w.cfg,)) | |
| print("%d tensors, %d parameters" % (len(w.state_dict), w.n_params)) | |
| return 0 | |
| def _weights_from_nlm_blob(blob): | |
| n=int.from_bytes(blob[:4],"little");md=json.loads(blob[4:4+n]);data=blob[4+n:];W={};G=md["G"] | |
| for k,shape,kind,off,ln in md["tensors"]: | |
| count=int(np.prod(shape)) | |
| if kind in ("f16","f32"): | |
| W[k]=np.frombuffer(data,np.float16 if kind=="f16" else np.float32,count,off).astype(np.float32).reshape(shape).copy() | |
| else: | |
| r,c=shape;ng=(c+G-1)//G | |
| if kind=="q8":q=np.frombuffer(data,np.int8,r*c,off).astype(np.float32).reshape(r,c);end=off+r*c | |
| else: | |
| nb=(r*c+1)//2;u=np.frombuffer(data,np.uint8,nb,off);q=(np.stack([u>>4,u&15],1).reshape(-1)[:r*c].astype(np.int16)-8).astype(np.float32).reshape(r,c);end=off+nb | |
| end+=(-end)%4;sc=np.frombuffer(data,np.float16,r*ng,end).astype(np.float32).reshape(r,ng);W[k]=q*np.repeat(sc,G,1)[:,:c] | |
| return md["cfg"],W | |
| def load_word_model(path): | |
| import torch | |
| from bhc.xlit_model import XlitTransformer, set_vocab | |
| w=load(path);set_vocab(w.cfg.get("vocab","hi"));m=XlitTransformer(**w.cfg);m.load_state_dict({k:torch.from_numpy(v.copy()) for k,v in w.state_dict.items()});return m.eval() | |
| def load_sentence_lm(path): | |
| from bhc.nlm import NLM | |
| w=load(path);vocab=json.loads(w.metadata["aksharalp.tokenizer_json"]) | |
| return NLM(w.cfg,w.state_dict,vocab,w.cfg.get("norm","self")) | |
| if __name__ == "__main__": | |
| sys.exit(main(sys.argv)) | |