From ead85305997c38a09723f0531d721c33cab69e55 Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Tue, 20 Oct 2026 00:37:06 +0300 Subject: [PATCH 1/2] validator: GGUF models on llama.cpp for the CPU groups, mt envelope check, and quantised weights in provenance --- docs/miner_setup.md | 2 +- microtensor_subnet/artifact/contract.py | 3 +- microtensor_subnet/artifact/gguf.py | 206 ++++++++++++++++++ microtensor_subnet/artifact/params.py | 28 ++- microtensor_subnet/artifact/validate.py | 54 ++++- microtensor_subnet/cli/envelope.py | 193 ++++++++++++++++ microtensor_subnet/cli/main.py | 2 + microtensor_subnet/validator/collect.py | 13 +- microtensor_subnet/validator/config.py | 2 + microtensor_subnet/validator/epoch.py | 57 ++++- microtensor_subnet/validator/items.py | 7 +- microtensor_subnet/validator/provenance.py | 147 +++++++++++-- .../validator/sandbox/__init__.py | 50 +++++ .../validator/sandbox/docker.py | 6 + microtensor_subnet/validator/sandbox/slm.py | 26 ++- pyproject.toml | 1 + 16 files changed, 756 insertions(+), 41 deletions(-) create mode 100644 microtensor_subnet/artifact/gguf.py create mode 100644 microtensor_subnet/cli/envelope.py diff --git a/docs/miner_setup.md b/docs/miner_setup.md index e542ff5..c07e35b 100644 --- a/docs/miner_setup.md +++ b/docs/miner_setup.md @@ -14,6 +14,6 @@ Validators score your weights under the arena's own prompt, and provenance fails 2. Harness: set the adapter prompt, the verifier and retries, measure again. 3. Distil: frontier answers with reasoning on schemas and question shapes like the arena's, filtered by the grader, fine tuned from the base (LoRA or full) and recorded under `training/`. 4. Reinforce: reward a query that returns the reference result on every database state; or preference pairs of a right teacher answer against a wrong draft. -5. Check: hold out whole domains, rename and revalue as the twin rule does, measure latency and memory inside your arena's envelope group (https://www.microtensor.cloud/docs/certified-inference/system/envelope-groups), train the router on the result. +5. Check: hold out whole domains, rename and revalue as the twin rule does, measure latency and memory inside your arena's envelope group with `mt envelope check --arena --slm ./slm_repo --items ` (https://www.microtensor.cloud/docs/certified-inference/system/envelope-groups), train the router on the result. Items are fresh every epoch and twins deduct memorisation, so train the skill, not the sample databases. Full page: https://www.microtensor.cloud/docs/certified-inference/guides/training diff --git a/microtensor_subnet/artifact/contract.py b/microtensor_subnet/artifact/contract.py index 07589f2..4ceecf5 100644 --- a/microtensor_subnet/artifact/contract.py +++ b/microtensor_subnet/artifact/contract.py @@ -30,12 +30,13 @@ "tokenizer.json", "tokenizer_config.json", ) +SLM_GGUF_REQUIRED_FILES: Final[tuple[str, ...]] = ("README.md", "LICENSE") SLM_TRAINING_FILES: Final[tuple[str, ...]] = ( "training/recipe.json", "training/data/sources.json", ) SLM_METRICS_GLOB: Final[str] = "training/runs/*/metrics.jsonl" -SLM_MIN_PARAMETERS: Final[int] = 1_000_000_000 +SLM_MIN_PARAMETERS: Final[int] = 100_000_000 SLM_MAX_PARAMETERS: Final[int] = 8_000_000_000 ROUTER_MAX_PARAMETERS: Final[int] = 1_500_000_000 ROUTER_CPU_PARAMETERS: Final[int] = 100_000_000 diff --git a/microtensor_subnet/artifact/gguf.py b/microtensor_subnet/artifact/gguf.py new file mode 100644 index 0000000..19d1856 --- /dev/null +++ b/microtensor_subnet/artifact/gguf.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import math +import struct +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, BinaryIO, Final + +MAGIC: Final[bytes] = b"GGUF" +KV_BYTES: Final[int] = 2 +SCALARS: Final[dict[int, str]] = { + 0: " int: + return math.prod(self.shape) if self.shape else 1 + + +@dataclass(slots=True) +class Gguf: + version: int + metadata: dict[str, Any] = field(default_factory=dict) + tensors: list[Tensor] = field(default_factory=list) + + @property + def parameters(self) -> int: + shapes = {t.name: t.shape for t in self.tensors} + tied = "output.weight" in shapes and shapes["output.weight"] == shapes.get( + "token_embd.weight" + ) + return sum(t.elements for t in self.tensors if not (tied and t.name == "output.weight")) + + @property + def file_type(self) -> str: + code = self.metadata.get("general.file_type") + return FILE_TYPES.get(int(code), str(code)) if code is not None else "" + + @property + def below_four_bits(self) -> list[str]: + return [t.name for t in self.tensors if t.kind in BELOW_FOUR_BITS] + + def number(self, suffix: str) -> int: + arch = str(self.metadata.get("general.architecture") or "") + value = self.metadata.get(f"{arch}.{suffix}") + return int(value) if isinstance(value, (int, float)) else 0 + + def kv_cache_bytes(self, max_len: int) -> int: + layers = self.number("block_count") + heads = self.number("attention.head_count") + kv_heads = self.number("attention.head_count_kv") or heads + width = self.number("embedding_length") + head_dim = self.number("attention.key_length") or (width // heads if heads else 0) + return 2 * layers * kv_heads * head_dim * max_len * KV_BYTES + + +def _unpack(handle: BinaryIO, fmt: str) -> Any: + size = struct.calcsize(fmt) + raw = handle.read(size) + if len(raw) != size: + raise GgufError("the GGUF header ends early") + return struct.unpack(fmt, raw)[0] + + +def _string(handle: BinaryIO) -> str: + length = _unpack(handle, " Any: + if kind in SCALARS: + return _unpack(handle, SCALARS[kind]) + if kind == STRING: + return _string(handle) + if kind == ARRAY: + inner = _unpack(handle, " Gguf: + with path.open("rb") as handle: + if handle.read(4) != MAGIC: + raise GgufError(f"{path.name} is not a GGUF file") + version = _unpack(handle, " list[Path]: + return sorted(p for p in root.glob("*.gguf") if p.is_file()) diff --git a/microtensor_subnet/artifact/params.py b/microtensor_subnet/artifact/params.py index 48b7299..92e0c93 100644 --- a/microtensor_subnet/artifact/params.py +++ b/microtensor_subnet/artifact/params.py @@ -7,6 +7,8 @@ from pathlib import Path from typing import Any +from microtensor_subnet.artifact import gguf + HEADER_PREFIX_BYTES = 8 QUANT_SUFFIXES = (".qweight", ".qzeros", ".scales", ".g_idx") INT4_PACK = 8 @@ -92,6 +94,28 @@ def kv_cache_bytes(root: Path, max_len: int) -> int: return 2 * layers * kv_heads * head_dim * max_len * KV_BYTES +def model_parameters(root: Path) -> int: + weights = safetensors_in(root) + if weights: + return count_parameters(weights) + found = gguf.gguf_in(root) + if len(found) != 1: + raise ParameterError("the model needs safetensors weights or exactly one GGUF file") + try: + return gguf.read(found[0]).parameters + except (OSError, gguf.GgufError) as exc: + raise ParameterError(str(exc)) from exc + + def footprint_bytes(root: Path, max_len: int) -> int: - weights = sum(path.stat().st_size for path in safetensors_in(root)) - return weights + kv_cache_bytes(root, max_len) + weights = safetensors_in(root) + if weights: + return sum(path.stat().st_size for path in weights) + kv_cache_bytes(root, max_len) + found = gguf.gguf_in(root) + if not found: + raise ParameterError(f"no weights in {root.name}") + try: + header = gguf.read(found[0]) + except (OSError, gguf.GgufError) as exc: + raise ParameterError(str(exc)) from exc + return found[0].stat().st_size + header.kv_cache_bytes(max_len) diff --git a/microtensor_subnet/artifact/validate.py b/microtensor_subnet/artifact/validate.py index 4343e25..5b68266 100644 --- a/microtensor_subnet/artifact/validate.py +++ b/microtensor_subnet/artifact/validate.py @@ -8,8 +8,13 @@ from pathlib import Path from typing import Any -from microtensor_subnet.artifact import contract -from microtensor_subnet.artifact.params import ParameterError, count_parameters, safetensors_in +from microtensor_subnet.artifact import contract, gguf +from microtensor_subnet.artifact.params import ( + ParameterError, + count_parameters, + model_parameters, + safetensors_in, +) _SHA = re.compile(r"^(sha256:)?[0-9a-f]{64}$") _GIT = re.compile(r"^[0-9a-f]{40}$") @@ -275,10 +280,22 @@ def _looks_secret(path: Path) -> bool: return any(marker in head for marker in _SECRET_MARKERS) -def check_slm_repo(root: Path, pin: SlmPin | None, allowed_bases: Iterable[str] = ()) -> Findings: +def check_slm_repo( + root: Path, + pin: SlmPin | None, + allowed_bases: Iterable[str] = (), + *, + bounds: tuple[int, int] | None = None, + formats: Iterable[str] = (), +) -> Findings: findings = Findings() present = set(tree_files(root)) - for name in contract.SLM_REQUIRED_FILES: + weights = safetensors_in(root) + ggufs = gguf.gguf_in(root) + required = ( + contract.SLM_GGUF_REQUIRED_FILES if ggufs and not weights else contract.SLM_REQUIRED_FILES + ) + for name in required: if name not in present: findings.problems.append(f"{name} is missing from the slm repository") for name in contract.SLM_TRAINING_FILES: @@ -286,20 +303,21 @@ def check_slm_repo(root: Path, pin: SlmPin | None, allowed_bases: Iterable[str] findings.problems.append(f"{name} is required for the archive") if not list(root.glob(contract.SLM_METRICS_GLOB)): findings.problems.append("training/runs//metrics.jsonl is required for the archive") - weights = safetensors_in(root) - if not weights: - findings.problems.append("no model.safetensors (or shards) in the slm repository root") + low, high = bounds or (contract.SLM_MIN_PARAMETERS, contract.SLM_MAX_PARAMETERS) + if not weights and not ggufs: + findings.problems.append("no model.safetensors (or shards) or model.gguf in the root") else: try: - parameters = count_parameters(weights) + parameters = model_parameters(root) findings.notes.append(f"slm parameters {parameters}") - if not contract.SLM_MIN_PARAMETERS <= parameters <= contract.SLM_MAX_PARAMETERS: + if not low <= parameters <= high: findings.problems.append( - f"slm has {parameters} parameters; the arena allows " - f"{contract.SLM_MIN_PARAMETERS} to {contract.SLM_MAX_PARAMETERS}" + f"slm has {parameters} parameters; the arena allows {low} to {high}" ) except ParameterError as exc: findings.problems.append(str(exc)) + if ggufs and not weights: + findings.problems += _gguf_problems(ggufs[0], [f.upper() for f in formats]) if pin is not None: for name, digest in pin.files.items(): if name not in present: @@ -318,6 +336,20 @@ def check_slm_repo(root: Path, pin: SlmPin | None, allowed_bases: Iterable[str] return findings +def _gguf_problems(path: Path, formats: list[str]) -> list[str]: + try: + header = gguf.read(path) + except (OSError, gguf.GgufError) as exc: + return [str(exc)] + problems: list[str] = [] + low = header.below_four_bits + if low: + problems.append(f"{len(low)} tensors are below 4 bit, for example {low[0]}") + if formats and header.file_type.upper() not in formats: + problems.append(f"GGUF type {header.file_type or 'unknown'} is not one of {formats}") + return problems + + def build_manifest( root: Path, *, diff --git a/microtensor_subnet/cli/envelope.py b/microtensor_subnet/cli/envelope.py new file mode 100644 index 0000000..b83a631 --- /dev/null +++ b/microtensor_subnet/cli/envelope.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import argparse +import json +import secrets +import time +import urllib.error +import urllib.request +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from microtensor_subnet.artifact.params import ParameterError, footprint_bytes, model_parameters +from microtensor_subnet.cli.common import add_common_arguments +from microtensor_subnet.core.constants import PUBLIC_SERVER_URL +from microtensor_subnet.scoring.stats import percentile +from microtensor_subnet.validator.config import LLAMA_IMAGE, VLLM_IMAGE +from microtensor_subnet.validator.context import render_prompt +from microtensor_subnet.validator.sandbox import Limits, cpu_container +from microtensor_subnet.validator.sandbox.docker import ContainerSpec, Docker, DockerError +from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError, vllm_command + +PORT = 18000 +READY_SECONDS = 900 +TIMEOUT_SECONDS = 20 +FILLER = "the agent opened the file, read the traceback and ran the tests again. " +CHARS_PER_TOKEN = 4 + + +def register(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None: + parser = subparsers.add_parser("envelope", help="measure a model against an arena's envelope") + inner = parser.add_subparsers(dest="action", required=True) + check = inner.add_parser("check", help="run the model under the validators' limits and time it") + check.add_argument("--arena", required=True) + check.add_argument("--slm", type=Path, required=True, help="slm repository root") + check.add_argument("--items", type=Path, help="items jsonl with a task per line") + check.add_argument("--count", type=int, default=20, help="prompts to time") + check.add_argument("--max-tokens", type=int, default=128, help="tokens to generate per prompt") + check.add_argument("--gpu", default="0", help="GPU index for GPU groups") + check.add_argument("--image", default="", help="override the vLLM or llama.cpp image") + check.add_argument("--port", type=int, default=PORT) + check.add_argument("--server", default=PUBLIC_SERVER_URL) + add_common_arguments(check) + check.set_defaults(handler=_check) + + +def _arena(server: str, arena_id: str) -> dict[str, Any]: + url = f"{server.rstrip('/')}/v2/arenas/{arena_id}" + try: + with urllib.request.urlopen(url, timeout=TIMEOUT_SECONDS) as answer: # noqa: S310 + found = json.loads(answer.read().decode("utf-8")) + except (urllib.error.URLError, ValueError) as exc: + raise SystemExit(f"arena {arena_id} could not be read from {server}: {exc}") from exc + return dict(found) + + +def _prompts( + definition: Mapping[str, Any], items: Path | None, count: int, length: int +) -> list[str]: + template = str(definition.get("slm_prompt") or "{user}") + if items is not None: + rows = [ + json.loads(line) + for line in items.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + return [render_prompt(template, dict(row.get("task") or {})) for row in rows[:count]] + filler = (FILLER * (length * CHARS_PER_TOKEN // len(FILLER) + 1))[: length * CHARS_PER_TOKEN] + return [render_prompt(template, {"system": "", "user": filler})] * count + + +def _spec( + args: argparse.Namespace, definition: Mapping[str, Any], limits: Limits, length: int +) -> ContainerSpec: + name = f"mt-check-{secrets.token_hex(3)}" + ports = ((f"127.0.0.1:{args.port}", 8000),) + if limits.on_cpu: + image = args.image or LLAMA_IMAGE + return cpu_container(name, image, args.slm.resolve(), length, limits, "bridge", ports) + return ContainerSpec( + name=name, + image=args.image or VLLM_IMAGE, + network="bridge", + command=vllm_command("/model", max_model_len=length), + env={"HF_HUB_OFFLINE": "1", "VLLM_NO_USAGE_STATS": "1"}, + volumes=((str(args.slm.resolve()), "/model", "ro"),), + gpus=(args.gpu,), + shm_size="8g", + ports=ports, + ) + + +def _wait(client: SlmClient, docker: Docker, name: str) -> None: + deadline = time.monotonic() + READY_SECONDS + while time.monotonic() < deadline: + if client.ready(): + return + if not docker.running(name): + raise SystemExit(f"the model server exited: {docker.logs(name, 20)[-600:]}") + time.sleep(5.0) + raise SystemExit("the model server did not become ready") + + +def _row(label: str, measured: str, limit: str, ok: bool) -> tuple[str, str, str, str]: + return label, measured, limit, "pass" if ok else "FAIL" + + +def _check(args: argparse.Namespace) -> int: + arena = _arena(args.server, args.arena) + definition = dict(arena.get("definition") or {}) + budgets = dict(definition.get("budgets") or {}) + max_input = int(budgets.get("max_input_tokens") or 0) + max_output = int(budgets.get("max_output_tokens") or 0) + if not max_input or not max_output: + raise SystemExit(f"arena {args.arena} names no envelope group") + length = max_input + max_output + limits = Limits.from_definition(definition, length) + prompts = _prompts(definition, args.items, args.count, max_input) + generation = definition.get("generation") or {} + stop = [str(s) for s in generation.get("stop") or []] + docker = Docker() + spec = _spec(args, definition, limits, length) + try: + docker.start(spec) + except DockerError as exc: + raise SystemExit(str(exc)) from exc + ttft: list[float] = [] + tpot: list[float] = [] + try: + client = SlmClient(f"http://127.0.0.1:{args.port}") + _wait(client, docker, spec.name) + for prompt in prompts: + try: + made = client.generate( + prompt, max_tokens=min(args.max_tokens, max_output), stop=stop + ) + except SlmError as exc: + raise SystemExit(str(exc)) from exc + ttft.append(made.ttft_ms) + if made.completion_tokens > 1: + tpot.append(made.tpot_ms) + memory = int(docker.usage(spec.name).get("memory_peak_bytes") or 0) if limits.on_cpu else 0 + finally: + docker.kill(spec.name) + try: + parameters = model_parameters(args.slm) + memory = memory or footprint_bytes(args.slm, length) + except ParameterError as exc: + raise SystemExit(str(exc)) from exc + rows = [ + _row( + "parameters", + f"{parameters:,}", + f"{int(budgets.get('min_parameters') or 0):,} to {int(budgets['max_parameters']):,}", + int(budgets.get("min_parameters") or 0) <= parameters <= int(budgets["max_parameters"]), + ), + _row( + "memory", + f"{memory / 1024**3:.2f} GB", + f"{budgets['memory_gb']} GB", + memory <= int(budgets["memory_bytes"]), + ), + ] + first = percentile(ttft, 0.95) + rows.append( + _row( + "first token p95", + f"{first:,.0f} ms", + f"{budgets['ttft_ms']:,} ms", + first <= budgets["ttft_ms"], + ) + ) + per = percentile(tpot, 0.95) + if budgets.get("tpot_ms"): + rows.append( + _row( + "per token p95", + f"{per:.1f} ms", + f"{budgets['tpot_ms']} ms", + per <= budgets["tpot_ms"], + ) + ) + if budgets.get("min_tokens_per_s"): + rate = 1000.0 / per if per > 0 else 0.0 + floor = float(budgets["min_tokens_per_s"]) + rows.append( + _row("generation", f"{rate:.1f} tokens/s", f"{floor:g} tokens/s", rate >= floor) + ) + group = (definition.get("envelope") or {}).get("group", "") + print(f"{args.arena} on {group}, {len(prompts)} prompts at up to {max_input:,} input tokens") + for label, measured, limit, verdict in rows: + print(f" {label:<17} {measured:>16} limit {limit:<26} {verdict}") + return 0 if all(r[3] == "pass" for r in rows) else 1 diff --git a/microtensor_subnet/cli/main.py b/microtensor_subnet/cli/main.py index 853cffd..9a79f65 100644 --- a/microtensor_subnet/cli/main.py +++ b/microtensor_subnet/cli/main.py @@ -3,6 +3,7 @@ import argparse import sys +from microtensor_subnet.cli import envelope as envelope_cmd from microtensor_subnet.cli import epoch as epoch_cmd from microtensor_subnet.cli import operator as operator_cmd from microtensor_subnet.cli import submit as submit_cmd @@ -25,6 +26,7 @@ def build_parser() -> argparse.ArgumentParser: submit_cmd.register(subparsers) epoch_cmd.register(subparsers) operator_cmd.register(subparsers) + envelope_cmd.register(subparsers) return parser diff --git a/microtensor_subnet/validator/collect.py b/microtensor_subnet/validator/collect.py index edef9a2..fbb1d6a 100644 --- a/microtensor_subnet/validator/collect.py +++ b/microtensor_subnet/validator/collect.py @@ -8,7 +8,10 @@ from typing import Any from microtensor_subnet.artifact import contract -from microtensor_subnet.artifact.params import ParameterError, count_parameters, safetensors_in +from microtensor_subnet.artifact.params import ( + ParameterError, + model_parameters, +) from microtensor_subnet.artifact.validate import ( ManifestError, check_artifact, @@ -116,6 +119,8 @@ def materialise( arena_id: str, allowed_bases: list[str], allowed_licenses: list[str], + bounds: tuple[int, int] | None = None, + formats: list[str] | None = None, ) -> Artifact: commitment = candidate.commitment try: @@ -140,13 +145,15 @@ def materialise( slm_dir = download(manifest.slm.repo, manifest.slm.revision, cache_dir) except Exception as exc: raise CollectError(f"slm repository could not be fetched: {exc}") from exc - more = check_slm_repo(slm_dir, manifest.slm, allowed_bases) + more = check_slm_repo( + slm_dir, manifest.slm, allowed_bases, bounds=bounds, formats=formats or () + ) findings.problems += more.problems licence = _licence(slm_dir) if allowed_licenses and licence not in [x.lower() for x in allowed_licenses]: findings.problems.append(f"slm licence {licence!r} is not allowed in this arena") try: - parameters = count_parameters(safetensors_in(slm_dir)) + parameters = model_parameters(slm_dir) except ParameterError as exc: findings.problems.append(str(exc)) if findings.problems: diff --git a/microtensor_subnet/validator/config.py b/microtensor_subnet/validator/config.py index 8eb3085..f0ac38b 100644 --- a/microtensor_subnet/validator/config.py +++ b/microtensor_subnet/validator/config.py @@ -8,6 +8,7 @@ from microtensor_subnet.core.constants import PUBLIC_SERVER_URL VLLM_IMAGE: Final[str] = "vllm/vllm-openai:v0.10.1" +LLAMA_IMAGE: Final[str] = "ghcr.io/ggml-org/llama.cpp:server" SANDBOX_IMAGE: Final[str] = "ghcr.io/microtensor-io/sandbox:1" GATEWAY_IMAGE: Final[str] = "ghcr.io/microtensor-io/validator:1" CONCURRENCY: Final[int] = 3 @@ -29,6 +30,7 @@ class EpochConfig: gpus: tuple[str, ...] = () concurrency: int = CONCURRENCY vllm_image: str = VLLM_IMAGE + llama_image: str = LLAMA_IMAGE sandbox_image: str = SANDBOX_IMAGE gateway_image: str = GATEWAY_IMAGE sandbox_runtime: str = "runsc" diff --git a/microtensor_subnet/validator/epoch.py b/microtensor_subnet/validator/epoch.py index fb68ac8..fc1f829 100644 --- a/microtensor_subnet/validator/epoch.py +++ b/microtensor_subnet/validator/epoch.py @@ -2,6 +2,7 @@ import json import logging +import platform import threading from collections.abc import Mapping from concurrent.futures import ThreadPoolExecutor @@ -150,6 +151,8 @@ def gather( arena_id=config.arena_id, allowed_bases=[str(b) for b in definition.get("allowed_bases") or []], allowed_licenses=[str(b) for b in definition.get("allowed_licenses") or []], + bounds=_bounds(definition), + formats=[str(f) for f in definition.get("allowed_formats") or []], ) except collect.CollectError as exc: failures[candidate.hotkey] = str(exc) @@ -183,6 +186,16 @@ def _memory(self, model_dir: Path, config: Mapping[str, Any]) -> int: log.warning("memory footprint of %s unknown: %s", model_dir.name, exc) return 0 + def _measured(self, sandbox: Sandbox, estimate: int) -> int: + if not self.limits.on_cpu: + return estimate + peak = int(self.host.docker.usage(sandbox.slm_container).get("memory_peak_bytes") or 0) + return peak or estimate + + @property + def limits(self) -> Limits: + return Limits.from_definition(self.context.definition, self.context.max_model_len) + def _open( self, artifact_dir: Path, @@ -204,7 +217,7 @@ def _open( slot=slot, databases=self.context.items.databases, item_databases=self.context.items.item_databases(), - limits=Limits.from_definition(self.context.definition, self.context.max_model_len), + limits=self.limits, ) def slm_phase(self, artifact: Artifact) -> None: @@ -228,6 +241,7 @@ def slm_phase(self, artifact: Artifact) -> None: memory_bytes=self._memory(artifact.slm_dir, artifact.config), parameters=artifact.slm_parameters, ) + outcome.slm.vram_peak_bytes = self._measured(sandbox, outcome.slm.vram_peak_bytes) finally: outcome.usage = self.host.close(sandbox) except (SandboxError, Exception) as exc: @@ -258,6 +272,9 @@ def reference_phase(self, base_dirs: Mapping[str, Path]) -> None: + self.context.items.items(ROUTER) + self.context.items.items(ESCALATION), ) + self.reference_slm.vram_peak_bytes = self._measured( + sandbox, self.reference_slm.vram_peak_bytes + ) if sandbox.router_url: self.reference_router = router.run( self.context, @@ -299,6 +316,9 @@ def reference_phase(self, base_dirs: Mapping[str, Path]) -> None: memory_bytes=self._memory(path, {}), parameters=0, ) + self.base_slm[name].vram_peak_bytes = self._measured( + base_sandbox, self.base_slm[name].vram_peak_bytes + ) finally: self.host.close(base_sandbox) finally: @@ -517,6 +537,7 @@ def run_epoch( relay = RelayClient(config.relay_url, str(token.get("token", ""))) desk = TicketDesk(relay, config.arena_id, epoch_index) engine = docker or Docker() + _check_host(context.definition) services = Services(host_ip(), config.host_services_port) gpus = tuple(config.gpus or ("0",)) agent: probe_judge.ProbeAgent | None = None @@ -533,8 +554,11 @@ def run_epoch( base_dirs[base] = collect.download(base, "main", config.cache_dir) except Exception as exc: log.warning("base %s could not be fetched: %s", base, exc) + measured_alone = runner.limits.on_cpu try: - with ThreadPoolExecutor(max_workers=max(1, config.concurrency)) as pool: + with ThreadPoolExecutor( + max_workers=1 if measured_alone else max(1, config.concurrency) + ) as pool: list(pool.map(runner.slm_phase, artifacts.values())) runner.reference_phase(base_dirs) peers = {h: o.slm.drafts() for h, o in runner.outcomes.items() if o.slm is not None} @@ -610,6 +634,35 @@ def run_epoch( ) +def _check_host(definition: Mapping[str, Any]) -> None: + envelope = definition.get("envelope") or {} + reference = envelope.get("reference") if isinstance(envelope, Mapping) else None + wanted = str((reference or {}).get("arch") or "") + if not wanted: + return + machine = platform.machine().lower() + host = ( + "arm64" + if machine in ("arm64", "aarch64") + else "x86_64" + if machine + in ( + "x86_64", + "amd64", + ) + else machine + ) + if host != wanted: + raise EpochError(f"this arena is measured on {wanted}; this host is {host}") + + +def _bounds(definition: Mapping[str, Any]) -> tuple[int, int] | None: + budgets = definition.get("budgets") or {} + if not isinstance(budgets, Mapping) or not budgets.get("max_parameters"): + return None + return int(budgets.get("min_parameters") or 0), int(budgets["max_parameters"]) + + def _owner(chain: Any) -> str: reader = getattr(chain, "owner_hotkey", None) if not callable(reader): diff --git a/microtensor_subnet/validator/items.py b/microtensor_subnet/validator/items.py index 947ab87..d1e4f0f 100644 --- a/microtensor_subnet/validator/items.py +++ b/microtensor_subnet/validator/items.py @@ -6,7 +6,10 @@ from pathlib import Path from typing import Any -from microtensor_subnet.artifact.params import ParameterError, count_parameters, safetensors_in +from microtensor_subnet.artifact.params import ( + ParameterError, + model_parameters, +) from microtensor_subnet.core.hashing import canonical_json from microtensor_subnet.core.items import ( ItemError, @@ -140,7 +143,7 @@ def reference_components( raise ReleaseError("the arena names no reference slm") slm_dir = download(slm_repo, slm_revision, cache_dir) try: - parameters = count_parameters(safetensors_in(slm_dir)) + parameters = model_parameters(slm_dir) except ParameterError as exc: raise ReleaseError(f"the reference slm has no readable weights: {exc}") from exc harness = trees.get("harness") diff --git a/microtensor_subnet/validator/provenance.py b/microtensor_subnet/validator/provenance.py index 3d159a4..357842e 100644 --- a/microtensor_subnet/validator/provenance.py +++ b/microtensor_subnet/validator/provenance.py @@ -7,12 +7,16 @@ from pathlib import Path from typing import Any +from microtensor_subnet.artifact.gguf import gguf_in from microtensor_subnet.artifact.params import read_header, safetensors_in from microtensor_subnet.artifact.validate import sha256_of log = logging.getLogger("microtensor_subnet.validator.provenance") ATTENTION_MARKERS = ("q_proj", "k_proj", "v_proj", "o_proj") +GGUF_NAMES = {"attn_q": "q_proj", "attn_k": "k_proj", "attn_v": "v_proj", "attn_output": "o_proj"} +AWQ_ORDER = (0, 4, 1, 5, 2, 6, 3, 7) +NIBBLE_SHIFTS = tuple(range(0, 32, 4)) SINGULAR_VALUES = 32 MAX_LAYERS = 8 DEFAULT_THRESHOLD = 0.9 @@ -36,38 +40,145 @@ def matches_public(slm_dir: Path, known: Iterable[str]) -> list[str]: wanted = set(known) if not wanted: return [] - return [path.name for path in safetensors_in(slm_dir) if sha256_of(path) in wanted] + files = safetensors_in(slm_dir) or gguf_in(slm_dir) + return [path.name for path in files if sha256_of(path) in wanted] -def _spectra(model_dir: Path, *, device: str = "cpu") -> dict[str, list[float]]: +def _normalised(matrix: Any) -> list[float]: + import numpy as np + + values = np.linalg.svd(np.asarray(matrix, dtype=np.float32), compute_uv=False)[:SINGULAR_VALUES] + scale = float(values[0]) if values.size and float(values[0]) > 0 else 1.0 + return [float(v) / scale for v in values] + + +def _unpack(packed: Any, axis: int) -> Any: + import numpy as np + + raw = np.asarray(packed).astype(np.uint32) + nibbles = np.stack([(raw >> shift) & 0xF for shift in NIBBLE_SHIFTS], axis=axis + 1) + shape = list(raw.shape) + shape[axis] *= len(NIBBLE_SHIFTS) + return nibbles.reshape(shape).astype(np.float32) + + +def dequantise_awq(qweight: Any, qzeros: Any, scales: Any, group: int) -> Any: + import numpy as np + + order = np.array(AWQ_ORDER) + + def unpacked(packed: Any) -> Any: + values = _unpack(packed, 1) + rows, cols = values.shape + return values.reshape(rows, cols // 8, 8)[:, :, order].reshape(rows, cols) + + weight = unpacked(qweight) + zeros = np.repeat(unpacked(qzeros), group, axis=0)[: weight.shape[0]] + factors = np.repeat(np.asarray(scales, dtype=np.float32), group, axis=0)[: weight.shape[0]] + return (weight - zeros) * factors + + +def dequantise_gptq( + qweight: Any, qzeros: Any, scales: Any, group: int, g_idx: Any | None = None +) -> Any: + import numpy as np + + weight = _unpack(qweight, 0) + zeros = _unpack(qzeros, 1) + 1 + rows = weight.shape[0] + index = np.asarray(g_idx) if g_idx is not None else np.arange(rows) // group + return (weight - zeros[index]) * np.asarray(scales, dtype=np.float32)[index] + + +def _quantisation(model_dir: Path) -> dict[str, Any]: + try: + config = json.loads((model_dir / "config.json").read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + found = config.get("quantization_config") if isinstance(config, Mapping) else None + return dict(found) if isinstance(found, Mapping) else {} + + +def _wanted(name: str, suffix: str) -> bool: + if not name.endswith(suffix) or not any(m in name for m in ATTENTION_MARKERS): + return False + layer = _layer_of(name) + return layer is not None and layer < MAX_LAYERS + + +def _safetensors_spectra(model_dir: Path) -> dict[str, list[float]]: try: import torch from safetensors import safe_open except ImportError as exc: raise RuntimeError("torch and safetensors are required for the weight fingerprint") from exc + + def array(tensor: Any) -> Any: + return (tensor if not tensor.is_floating_point() else tensor.to(torch.float32)).numpy() + + quant = _quantisation(model_dir) + method = str(quant.get("quant_method") or "").lower() + group = int(quant.get("group_size") or 128) found: dict[str, list[float]] = {} for path in safetensors_in(model_dir): header = read_header(path) - names = sorted( - name - for name, entry in header.items() - if isinstance(entry, Mapping) - and any(marker in name for marker in ATTENTION_MARKERS) - and name.endswith(".weight") - and len(entry.get("shape") or []) == 2 - ) - with safe_open(str(path), framework="pt", device=device) as handle: + names = sorted(n for n, e in header.items() if isinstance(e, Mapping)) + with safe_open(str(path), framework="pt", device="cpu") as handle: for name in names: - layer = _layer_of(name) - if layer is None or layer >= MAX_LAYERS: - continue - tensor = handle.get_tensor(name).to(torch.float32) - values = torch.linalg.svdvals(tensor)[:SINGULAR_VALUES] - scale = float(values[0]) if values.numel() and float(values[0]) > 0 else 1.0 - found[name] = [float(v) / scale for v in values.tolist()] + if _wanted(name, ".weight") and len(header[name].get("shape") or []) == 2: + found[name] = _normalised(array(handle.get_tensor(name))) + elif _wanted(name, ".qweight") and method in ("awq", "gptq"): + stem = name[: -len(".qweight")] + parts = { + key: array(handle.get_tensor(f"{stem}.{key}")) + for key in ("qzeros", "scales", "g_idx") + if f"{stem}.{key}" in header + } + packed = array(handle.get_tensor(name)) + matrix = ( + dequantise_awq(packed, parts["qzeros"], parts["scales"], group) + if method == "awq" + else dequantise_gptq( + packed, parts["qzeros"], parts["scales"], group, parts.get("g_idx") + ) + ) + found[f"{stem}.weight"] = _normalised(matrix) return found +def _gguf_name(name: str) -> str: + parts = name.split(".") + if len(parts) != 4 or parts[0] != "blk" or parts[3] != "weight" or not parts[1].isdigit(): + return "" + projection = GGUF_NAMES.get(parts[2], "") + return f"model.layers.{parts[1]}.self_attn.{projection}.weight" if projection else "" + + +def _gguf_spectra(path: Path) -> dict[str, list[float]]: + try: + from gguf import GGUFReader + from gguf.quants import dequantize + except ImportError as exc: + raise RuntimeError("the gguf package is required to fingerprint GGUF weights") from exc + found: dict[str, list[float]] = {} + for tensor in GGUFReader(str(path)).tensors: + name = _gguf_name(str(tensor.name)) + layer = _layer_of(name) if name else None + if layer is None or layer >= MAX_LAYERS: + continue + found[name] = _normalised(dequantize(tensor.data, tensor.tensor_type)) + return found + + +def _spectra(model_dir: Path) -> dict[str, list[float]]: + if safetensors_in(model_dir): + return _safetensors_spectra(model_dir) + ggufs = gguf_in(model_dir) + if ggufs: + return _gguf_spectra(ggufs[0]) + raise RuntimeError(f"no weights to fingerprint in {model_dir.name}") + + def _layer_of(name: str) -> int | None: for part in name.split("."): if part.isdigit(): diff --git a/microtensor_subnet/validator/sandbox/__init__.py b/microtensor_subnet/validator/sandbox/__init__.py index faeb4a1..d175a16 100644 --- a/microtensor_subnet/validator/sandbox/__init__.py +++ b/microtensor_subnet/validator/sandbox/__init__.py @@ -17,6 +17,7 @@ import httpx from microtensor_subnet.artifact import contract +from microtensor_subnet.artifact.gguf import gguf_in from microtensor_subnet.validator.relay import TicketDesk from microtensor_subnet.validator.sandbox import tools from microtensor_subnet.validator.sandbox.docker import ContainerSpec, Docker, DockerError @@ -25,6 +26,7 @@ MEMORY_SHARE, SHARED_MEMORY_SHARE, SlmClient, + llama_command, vllm_command, ) @@ -45,6 +47,9 @@ class SandboxConfig(Protocol): @property def vllm_image(self) -> str: ... + @property + def llama_image(self) -> str: ... + @property def sandbox_image(self) -> str: ... @@ -98,6 +103,13 @@ class Limits: router_max_input_tokens: int = ROUTER_INPUT_CEILING max_model_calls: int = 0 model_faults: bool = False + device: str = "gpu" + cpu_threads: int = 0 + memory_cap_gb: float = 0.0 + + @property + def on_cpu(self) -> bool: + return self.device == "cpu" @property def router_on_gpu(self) -> bool: @@ -108,6 +120,10 @@ def from_definition(cls, definition: Mapping[str, Any], max_model_len: int | Non limits = definition.get("limits") or {} router = limits.get("router") or {} if isinstance(limits, Mapping) else {} harness = limits.get("harness") or {} if isinstance(limits, Mapping) else {} + envelope = definition.get("envelope") or {} + envelope = envelope if isinstance(envelope, Mapping) else {} + reference = envelope.get("reference") or {} + reference = reference if isinstance(reference, Mapping) else {} return cls( max_model_len=max_model_len, router_max_parameters=int( @@ -116,9 +132,38 @@ def from_definition(cls, definition: Mapping[str, Any], max_model_len: int | Non router_max_input_tokens=int(router.get("max_input_tokens") or ROUTER_INPUT_CEILING), max_model_calls=int(harness.get("max_model_calls") or 0), model_faults=harness.get("faults") == "model", + device=str(envelope.get("device") or "gpu"), + cpu_threads=int(reference.get("threads") or 0), + memory_cap_gb=float(reference.get("memory_cap_gb") or 0), ) +def cpu_container( + name: str, + image: str, + model_dir: Path, + max_len: int, + limits: Limits, + network: str, + ports: Sequence[tuple[str, int]] = (), +) -> ContainerSpec: + found = gguf_in(model_dir) + if not found: + raise SandboxError("a CPU envelope needs one GGUF file in the model repository") + threads = max(1, limits.cpu_threads) + return ContainerSpec( + name=name, + image=image, + network=network, + command=llama_command(f"/model/{found[0].name}", threads=threads, max_model_len=max_len), + volumes=((str(model_dir), "/model", "ro"),), + cpus=float(threads), + cpuset=f"0-{threads - 1}", + memory=f"{limits.memory_cap_gb:g}g" if limits.memory_cap_gb else "", + ports=ports, + ) + + @dataclass(slots=True) class Sandbox: key: str @@ -156,6 +201,11 @@ def _slm_container( ) -> str: name = f"mt-slm-{key}" max_len = limits.max_model_len or int((config.get("slm") or {}).get("max_context") or 8192) + if limits.on_cpu: + self.docker.start( + cpu_container(name, self.config.llama_image, model_dir, max_len, limits, network) + ) + return name share = SHARED_MEMORY_SHARE if limits.router_on_gpu else MEMORY_SHARE self.docker.start( ContainerSpec( diff --git a/microtensor_subnet/validator/sandbox/docker.py b/microtensor_subnet/validator/sandbox/docker.py index 9f207fc..48d60ca 100644 --- a/microtensor_subnet/validator/sandbox/docker.py +++ b/microtensor_subnet/validator/sandbox/docker.py @@ -38,6 +38,8 @@ class ContainerSpec: tmpfs: Sequence[str] = () shm_size: str = "" ipc_none: bool = False + cpuset: str = "" + ports: Sequence[tuple[str, int]] = () class Runner(Protocol): @@ -71,6 +73,10 @@ def _flags(spec: ContainerSpec) -> list[str]: argv += ["--runtime", spec.runtime] if spec.cpus: argv += ["--cpus", f"{spec.cpus:g}"] + if spec.cpuset: + argv += ["--cpuset-cpus", spec.cpuset] + for host, port in spec.ports: + argv += ["-p", f"{host}:{port}"] if spec.memory: argv += ["--memory", spec.memory, "--memory-swap", spec.memory] if spec.pids: diff --git a/microtensor_subnet/validator/sandbox/slm.py b/microtensor_subnet/validator/sandbox/slm.py index 739e943..26e63b8 100644 --- a/microtensor_subnet/validator/sandbox/slm.py +++ b/microtensor_subnet/validator/sandbox/slm.py @@ -62,6 +62,7 @@ def generate(self, prompt: str, *, max_tokens: int, stop: Sequence[str] = ()) -> started = time.perf_counter() first: float | None = None text: list[str] = [] + pieces = 0 usage: dict[str, Any] = {} finish = "" try: @@ -89,6 +90,8 @@ def generate(self, prompt: str, *, max_tokens: int, stop: Sequence[str] = ()) -> piece = str(choice.get("text") or "") if piece and first is None: first = time.perf_counter() + if piece: + pieces += 1 text.append(piece) if choice.get("finish_reason"): finish = str(choice["finish_reason"]) @@ -98,13 +101,34 @@ def generate(self, prompt: str, *, max_tokens: int, stop: Sequence[str] = ()) -> return Generation( text="".join(text), prompt_tokens=int(usage.get("prompt_tokens") or 0), - completion_tokens=int(usage.get("completion_tokens") or 0), + completion_tokens=int(usage.get("completion_tokens") or pieces), ttft_ms=((first or ended) - started) * 1000, total_ms=(ended - started) * 1000, finish_reason=finish, ) +def llama_command(model_file: str, *, threads: int, max_model_len: int) -> list[str]: + return [ + "-m", + model_file, + "--host", + "0.0.0.0", # noqa: S104 + "--port", + "8000", + "--alias", + MODEL_NAME, + "-t", + str(threads), + "-tb", + str(threads), + "-c", + str(max_model_len), + "-np", + "1", + ] + + def vllm_command( model_dir: str, *, diff --git a/pyproject.toml b/pyproject.toml index 42e5b75..de7f77d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,7 @@ dependencies = [ [project.optional-dependencies] validator = [ + "gguf>=0.10", "torch>=2.4", "transformers>=4.44", "safetensors>=0.4", From 5485bcf59fc3215a15457a847f08d8beef71b9bd Mon Sep 17 00:00:00 2001 From: BANADDA MUBARAKA <83466862+BANADDA@users.noreply.github.com> Date: Tue, 20 Oct 2026 00:40:48 +0300 Subject: [PATCH 2/2] mypy: treat gguf as an untyped optional import --- pyproject.toml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index de7f77d..6edef16 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,6 +110,8 @@ module = [ "boto3.*", "botocore.*", "huggingface_hub.*", + "gguf", + "gguf.*", "llama_cpp.*", "onnx.*", "onnxruntime.*",