From 7183e6de8c6fcb5c62241f3363ea1234ee21f52b Mon Sep 17 00:00:00 2001 From: George Kim Date: Fri, 5 Jun 2026 16:20:53 +0000 Subject: [PATCH] feat(owner_eval): standalone owner eval pipeline (GSM8K/MMLU/PPL -> Prometheus) --- connito/owner_eval/OBSERVABILITY.md | 200 +++++++++++++++++++++ connito/owner_eval/__init__.py | 12 ++ connito/owner_eval/bootstrap.py | 112 ++++++++++++ connito/owner_eval/cli.py | 9 + connito/owner_eval/metrics/__init__.py | 8 + connito/owner_eval/metrics/base.py | 126 +++++++++++++ connito/owner_eval/metrics/gsm8k_ppl.py | 36 ++++ connito/owner_eval/metrics/gsm8k_task.py | 58 ++++++ connito/owner_eval/metrics/mmlu.py | 55 ++++++ connito/owner_eval/registry.py | 64 +++++++ connito/owner_eval/run.py | 142 +++++++++++++++ connito/owner_eval/runner.py | 46 +++++ connito/owner_eval/text_utils.py | 52 ++++++ connito/shared/config.py | 51 ++++++ connito/shared/telemetry.py | 66 +++++++ connito/test/test_owner_eval_bootstrap.py | 59 ++++++ connito/test/test_owner_eval_metrics.py | 185 +++++++++++++++++++ connito/test/test_owner_eval_registry.py | 84 +++++++++ connito/test/test_owner_eval_runner.py | 53 ++++++ connito/test/test_owner_eval_text_utils.py | 47 +++++ observability/prometheus.yml | 7 + owner_eval.canary.yaml | 31 ++++ pyproject.toml | 1 + 23 files changed, 1504 insertions(+) create mode 100644 connito/owner_eval/OBSERVABILITY.md create mode 100644 connito/owner_eval/__init__.py create mode 100644 connito/owner_eval/bootstrap.py create mode 100644 connito/owner_eval/cli.py create mode 100644 connito/owner_eval/metrics/__init__.py create mode 100644 connito/owner_eval/metrics/base.py create mode 100644 connito/owner_eval/metrics/gsm8k_ppl.py create mode 100644 connito/owner_eval/metrics/gsm8k_task.py create mode 100644 connito/owner_eval/metrics/mmlu.py create mode 100644 connito/owner_eval/registry.py create mode 100644 connito/owner_eval/run.py create mode 100644 connito/owner_eval/runner.py create mode 100644 connito/owner_eval/text_utils.py create mode 100644 connito/test/test_owner_eval_bootstrap.py create mode 100644 connito/test/test_owner_eval_metrics.py create mode 100644 connito/test/test_owner_eval_registry.py create mode 100644 connito/test/test_owner_eval_runner.py create mode 100644 connito/test/test_owner_eval_text_utils.py create mode 100644 owner_eval.canary.yaml diff --git a/connito/owner_eval/OBSERVABILITY.md b/connito/owner_eval/OBSERVABILITY.md new file mode 100644 index 0000000..9d7cc52 --- /dev/null +++ b/connito/owner_eval/OBSERVABILITY.md @@ -0,0 +1,200 @@ +# Owner Eval — Backend API Consumption Guide + +How the backend / leaderboard API surfaces the owner eval pipeline's results. + +## Architecture (pull model) + +``` + owner_eval daemon central Prometheus backend API / dashboard + ┌────────────────┐ scrape /metrics ┌──────────────┐ PromQL over HTTP ┌──────────────┐ + │ exposes :8400 │ ◀──────────────── │ job: │ ◀───────────────── │ /api/v1/query│ + │ owner_eval_* │ every 5–15s │ mycelia_ │ │ render panels│ + │ gauges/counter │ │ owner_eval │ └──────────────┘ + └────────────────┘ └──────────────┘ +``` + +The daemon does **not** push anywhere and the backend does **not** talk to the daemon +directly. The daemon exposes a Prometheus text endpoint on `eval_pipeline.telemetry_port` +(default **8400**); Prometheus scrapes it; the backend queries **Prometheus**, never the +daemon. This mirrors how validator/miner metrics already flow. + +### Prometheus must scrape the daemon + +Add the daemon as a target (already in `observability/prometheus.yml` for local dev; the +production Prometheus needs the same target pointed at the daemon's real host): + +```yaml + - job_name: 'mycelia_owner_eval' + static_configs: + - targets: [':8400'] # eval_pipeline.telemetry_port +``` + +Until this target exists and reports `up`, the backend cannot see any owner_eval data. + +## Metric reference (the contract) + +Every series also carries Prometheus-added labels `job="mycelia_owner_eval"` and +`instance=":8400"`. + +| Series | Type | Labels | Meaning | +|---|---|---|---| +| `owner_eval_metric` | Gauge | `metric` | One scalar result per run, keyed by `metric`. Values below. | +| `owner_eval_status` | Gauge | `metric` (evaluator name) | `1` = evaluator's last run succeeded, `0` = it raised. | +| `owner_eval_run_info` | Gauge (=1) | `model_revision`, `cycle_index` | Identity of the latest run. Join target for context. | +| `owner_eval_last_run_timestamp` | Gauge | — | Unix seconds of the last completed run (freshness). | +| `owner_eval_loop_heartbeat_total` | Counter | — | Daemon poll iterations (liveness). | + +**`owner_eval_metric` keys** (the `metric` label) — one Prometheus sample each: + +| `metric` | Meaning | Range | +|---|---|---| +| `gsm8k_task_acc` | GSM8K exact-match accuracy | 0–1 | +| `gsm8k_task_n` | GSM8K samples evaluated | int | +| `mmlu_acc` | MMLU accuracy | 0–1 | +| `mmlu_n` | MMLU samples evaluated | int | +| `gsm8k_ppl` | GSM8K perplexity (`exp(val_loss)`) | ≥1, lower better | +| `gsm8k_ppl_val_loss` | mean LM loss behind the ppl | ≥0 | + +New metrics appear automatically as new `metric` label values — the backend should treat +the key set as **open** (render whatever keys are present) rather than hard-coding it. + +**`owner_eval_status` keys** are the *evaluator* names — `gsm8k_ppl`, `gsm8k_task`, `mmlu` +— not the per-scalar keys above (one evaluator can emit several scalars). + +## Querying Prometheus (what the backend calls) + +Prometheus HTTP API, typically `http://:9090`: + +- **Instant** (current value): `GET /api/v1/query?query=` +- **Range** (history/trend): `GET /api/v1/query_range?query=&start=&end=&step=` +- **Targets health**: `GET /api/v1/targets?state=active` + +### Response shape (instant query) + +```json +{ + "status": "success", + "data": { + "resultType": "vector", + "result": [ + { "metric": {"__name__":"owner_eval_metric","metric":"mmlu_acc", + "job":"mycelia_owner_eval","instance":"host:8400"}, + "value": [1780671984.769, "0.5"] } + ] + } +} +``` + +`value` is `[, ""]` — parse the second element as a float. + +### Latest leaderboard values + +``` +owner_eval_metric +``` +Returns all current scalars in one call; group by the `metric` label to populate the board. + +### Attach model revision + cycle to each value (instant) + +``` +owner_eval_metric + * on(instance) group_left(model_revision, cycle_index) owner_eval_run_info +``` +Multiplies by `owner_eval_run_info` (always `1`) and grafts its `model_revision` / +`cycle_index` labels onto every metric — so each displayed score is tagged with which model +produced it. + +### Trend over time (range query) + +``` +owner_eval_metric{metric="mmlu_acc"} +``` +`owner_eval_metric` has **stable labels**, so a `query_range` gives a clean time series per +metric for charting progress across runs. + +### Health / freshness / liveness + +``` +up{job="mycelia_owner_eval"} # 1 = Prometheus is scraping the daemon +time() - owner_eval_last_run_timestamp # seconds since last completed run (staleness) +owner_eval_status # per-evaluator 1/0 success +rate(owner_eval_loop_heartbeat_total[10m]) > 0 # daemon poll loop alive +``` + +Suggested alerts: `up == 0` (daemon/scrape down), `time() - owner_eval_last_run_timestamp > +6 * 3600` (no run in ~last interval window), `owner_eval_status == 0` (an evaluator failing), +`rate(owner_eval_loop_heartbeat_total[15m]) == 0` (daemon hung). + +### Example calls + +```bash +# current scores, tagged with model revision + cycle +curl -G 'http://prometheus:9090/api/v1/query' \ + --data-urlencode 'query=owner_eval_metric * on(instance) group_left(model_revision,cycle_index) owner_eval_run_info' + +# mmlu accuracy over the last 30 days, daily points +curl -G 'http://prometheus:9090/api/v1/query_range' \ + --data-urlencode 'query=owner_eval_metric{metric="mmlu_acc"}' \ + --data-urlencode "start=$(date -d -30days +%s)" \ + --data-urlencode "end=$(date +%s)" \ + --data-urlencode 'step=86400' +``` + +```python +import requests + +PROM = "http://prometheus:9090" + +def latest_scores(): + q = ('owner_eval_metric * on(instance) ' + 'group_left(model_revision,cycle_index) owner_eval_run_info') + r = requests.get(f"{PROM}/api/v1/query", params={"query": q}, timeout=5).json() + out = [] + for s in r["data"]["result"]: + m = s["metric"] + out.append({ + "metric": m["metric"], + "value": float(s["value"][1]), + "model_revision": m.get("model_revision"), + "cycle_index": m.get("cycle_index"), + }) + return out +``` + +## Semantics & caveats + +- **Gauges are last-value.** `owner_eval_metric` is overwritten each run; "history" comes + from Prometheus's stored samples (retention), queried with `query_range`. There is no + separate history endpoint. +- **`owner_eval_run_info` is an info metric** (value always `1`; `model_revision` / + `cycle_index` carried as labels, replaced each run — bounded cardinality, one series at a + time). Use the `group_left` join for *current* context. It is **not** a reliable way to + reconstruct which revision produced a *historical* value (its labelset changes over time). + If the backend needs durable per-model-version history, **snapshot** `owner_eval_metric` + + `owner_eval_run_info` into its own store whenever `cycle_index` advances. (Alternatively + the pipeline could add `model_revision` as a label on `owner_eval_metric` itself, at the + cost of a new series per model version — ask if you want that.) +- **`model_revision`** is `globalver_` for chain-fetched models, or `base` when the + daemon runs in `model_source: base` (canary) mode — filter out `base` for production + boards if a canary is ever pointed at the same Prometheus. +- **n changes don't change series**; `gsm8k_task_n` / `mmlu_n` report the sample count each + run so the backend can show the confidence basis (e.g. "acc 0.41 over n=500"). + +## Reproduce locally (canary) + +```bash +# 1. run the daemon against the base model (no wallet), publishing tiny-n results +CUDA_VISIBLE_DEVICES=0 .venv/bin/python -m connito.owner_eval.run \ + --path owner_eval.canary.yaml # model_source: base, n=2, port 8400 + +# 2. confirm the publish side +curl -s localhost:8400/metrics | grep owner_eval + +# 3. point a Prometheus at it (observability/prometheus.yml already has the job) +docker run --rm -d --name prom --network host \ + -v "$PWD/observability/prometheus.yml:/etc/prometheus/prometheus.yml" prom/prometheus:latest + +# 4. query as the backend would +curl -s 'localhost:9090/api/v1/targets?state=active' | grep mycelia_owner_eval # => up +curl -s 'localhost:9090/api/v1/query?query=owner_eval_metric' +``` diff --git a/connito/owner_eval/__init__.py b/connito/owner_eval/__init__.py new file mode 100644 index 0000000..4062cb6 --- /dev/null +++ b/connito/owner_eval/__init__.py @@ -0,0 +1,12 @@ +"""Standalone owner-run evaluation pipeline. + +A daemon (``connito.owner_eval.run``) that the subnet owner runs independently of +any miner or validator. Every N cycles it downloads the latest merged full model +that a validator published to HuggingFace, runs a pluggable benchmark suite +(GSM8K perplexity, GSM8K task accuracy, MMLU accuracy to start), and emits the +results as Prometheus metrics for the leaderboard dashboard to scrape. + +New metrics are added by dropping an ``Evaluator`` subclass into +``connito/owner_eval/metrics/`` and registering it with ``@register`` — see +``connito.owner_eval.registry``. +""" diff --git a/connito/owner_eval/bootstrap.py b/connito/owner_eval/bootstrap.py new file mode 100644 index 0000000..4baa06f --- /dev/null +++ b/connito/owner_eval/bootstrap.py @@ -0,0 +1,112 @@ +"""Bootstrap + model-loading helpers for the owner eval daemon. + +The read-only config load, device resolution, precision handling and +model-release helpers are adapted from +``notebook/diagnose_eval_parity.py`` (the offline eval-parity diagnostic) so the +two share the same proven setup path. Loading the *latest validator HF +checkpoint* into the full model is owner-eval-specific and goes through +``connito.shared.model.load_model``. +""" + +from __future__ import annotations + +import gc +from typing import Any + +import torch +import yaml + +from connito.shared.app_logging import structlog + +logger = structlog.get_logger(__name__) + + +def load_config(config_path: str) -> Any: + """Load an ``OwnerEvalConfig`` from a YAML file (read-only).""" + from connito.shared.config import OwnerEvalConfig + + with open(config_path, encoding="utf-8") as fh: + data = yaml.safe_load(fh) or {} + return OwnerEvalConfig(**data) + + +def resolve_device(config: Any) -> torch.device: + configured = getattr(config.model, "device", None) + if configured: + if str(configured).startswith("cuda") and not torch.cuda.is_available(): + logger.warning("Configured CUDA device unavailable; falling back to CPU", + configured_device=configured) + return torch.device("cpu") + return torch.device(configured) + return torch.device("cuda" if torch.cuda.is_available() else "cpu") + + +def release_model(model: torch.nn.Module | None) -> None: + """Drop a model and reclaim GPU memory between runs.""" + if model is None: + return + del model + gc.collect() + if torch.cuda.is_available(): + torch.cuda.empty_cache() + + +def _model_dtype(config: Any) -> torch.dtype: + precision = getattr(config.model, "precision", "fp16-mixed") + if precision == "bf16-mixed" and torch.cuda.is_available() and not torch.cuda.is_bf16_supported(): + precision = "fp16-mixed" + return torch.bfloat16 if precision == "bf16-mixed" else torch.float16 + + +def load_base_model(config: Any, expert_manager: Any, device: torch.device) -> torch.nn.Module: + """Load the pretrained base full model directly (no chain/wallet). + + Used for canary / plumbing tests (``eval_pipeline.model_source == "base"``). + """ + from connito.shared.modeling.mycelia import get_base_model + + model = get_base_model(config, expert_manager=expert_manager, group_ids=None, partial=False) + model = model.to(device=device, dtype=_model_dtype(config)).eval() + return model + + +def load_latest_full_model( + config: Any, + expert_manager: Any, + subtensor: Any, + wallet: Any, + device: torch.device, +) -> tuple[torch.nn.Module, Any]: + """Build the full model under test. + + With ``eval_pipeline.model_source == "base"`` loads the pretrained base model + directly (canary mode). Otherwise wraps ``connito.shared.model.load_model`` + with ``partial=False`` and ``current_checkpoint=None`` (always take the newest + validator HF checkpoint). Returns ``(model, ModelCheckpoint | None)``; the + checkpoint's ``global_ver`` stamps the model revision into telemetry. + """ + if getattr(config.eval_pipeline, "model_source", "chain") == "base": + return load_base_model(config, expert_manager, device), None + + from connito.shared.model import load_model + + model, checkpoint = load_model( + rank=0, + config=config, + expert_manager=expert_manager, + subtensor=subtensor, + wallet=wallet, + current_checkpoint=None, + partial=False, + checkpoint_device=device, + ) + model.eval() + return model, checkpoint + + +def model_revision_label(checkpoint: Any) -> str: + """Human/Prometheus-friendly revision string for a loaded checkpoint.""" + global_ver = getattr(checkpoint, "global_ver", None) + if global_ver is None: + return "unknown" + return f"globalver_{global_ver}" diff --git a/connito/owner_eval/cli.py b/connito/owner_eval/cli.py new file mode 100644 index 0000000..455651f --- /dev/null +++ b/connito/owner_eval/cli.py @@ -0,0 +1,9 @@ +"""Console-script shim for the owner eval daemon (parity with miner/validator CLIs).""" + +from __future__ import annotations + +from connito.owner_eval.run import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/connito/owner_eval/metrics/__init__.py b/connito/owner_eval/metrics/__init__.py new file mode 100644 index 0000000..6f8c814 --- /dev/null +++ b/connito/owner_eval/metrics/__init__.py @@ -0,0 +1,8 @@ +"""Importing this package registers all built-in evaluators via ``@register``. + +Add a new metric module here so its registration side-effect runs. +""" + +from connito.owner_eval.metrics import gsm8k_ppl, gsm8k_task, mmlu # noqa: F401 + +__all__ = ["gsm8k_ppl", "gsm8k_task", "mmlu"] diff --git a/connito/owner_eval/metrics/base.py b/connito/owner_eval/metrics/base.py new file mode 100644 index 0000000..7139025 --- /dev/null +++ b/connito/owner_eval/metrics/base.py @@ -0,0 +1,126 @@ +"""Base evaluator + shared dataset/scoring helpers for owner eval metrics.""" + +from __future__ import annotations + +from typing import Any + +import torch + +from connito.shared.app_logging import structlog +from connito.owner_eval.text_utils import ( # re-exported for metric modules/tests + extract_final_number, + extract_gsm8k_gold, + numeric_eq, +) + +logger = structlog.get_logger(__name__) + +__all__ = [ + "BaseEvaluator", + "prep_tokenizer", + "load_hf_split", + "make_lm_batches", + "loglikelihood", + "extract_final_number", + "extract_gsm8k_gold", + "numeric_eq", +] + + +class BaseEvaluator: + """Common base for registered evaluators. + + Subclasses set ``name`` (the Prometheus metric label / registry key) and + implement ``evaluate``. ``default_n_samples`` is a per-class fallback used + when the metric isn't listed in ``config.eval_pipeline.n_samples_per_metric``. + """ + + name: str = "" + default_n_samples: int = 100 + + def n_samples(self, config: Any) -> int: + per_metric = getattr(config.eval_pipeline, "n_samples_per_metric", {}) or {} + return int(per_metric.get(self.name, getattr(config.eval_pipeline, "default_n_samples", self.default_n_samples))) + + def evaluate(self, model: Any, tokenizer: Any, device: Any, config: Any) -> dict[str, float]: + raise NotImplementedError + + +def prep_tokenizer(tokenizer): + """Ensure a pad token exists (DeepSeek's tokenizer ships without one).""" + if getattr(tokenizer, "pad_token", None) is None: + tokenizer.pad_token = tokenizer.eos_token + return tokenizer + + +def load_hf_split(path: str, name: str | None, split: str, n: int, seed: int | None = None): + """Load up to ``n`` rows of an HF dataset split (non-streaming; n is small). + + When ``seed`` is given, the split is shuffled with that fixed seed before + selecting ``n`` rows. Use this for datasets whose row order is itself + structured — e.g. ``cais/mmlu`` "all" is ordered by subject, so taking the + first n rows would only cover the alphabetically-earliest subjects. A seeded + shuffle yields a representative sample across subjects while staying + deterministic across runs. + """ + from datasets import load_dataset + + ds = load_dataset(path, name, split=split) if name else load_dataset(path, split=split) + if seed is not None: + ds = ds.shuffle(seed=seed) + n = min(n, len(ds)) + return ds.select(range(n)) + + +def make_lm_batches(texts, tokenizer, seq_len: int, batch_size: int): + """Tokenize ``texts`` into causal-LM batches compatible with ``evaluate_model``. + + Yields dicts of ``{input_ids, attention_mask, labels}`` (labels == input_ids + with pad positions masked to -100), the same shape the validator's + DataCollatorForLanguageModeling(mlm=False) produces. + """ + batches = [] + for start in range(0, len(texts), batch_size): + chunk = texts[start:start + batch_size] + enc = tokenizer( + chunk, + truncation=True, + max_length=seq_len, + padding=True, + return_tensors="pt", + ) + input_ids = enc["input_ids"] + attention_mask = enc["attention_mask"] + labels = input_ids.clone() + labels[attention_mask == 0] = -100 + batches.append({ + "input_ids": input_ids, + "attention_mask": attention_mask, + "labels": labels, + }) + return batches + + +@torch.no_grad() +def loglikelihood(model, tokenizer, prompt: str, continuation: str, device, normalize: bool = True) -> float: + """Teacher-forced log-likelihood of ``continuation`` given ``prompt``. + + Single forward pass; sums the log-probabilities the model assigns to the + continuation tokens. When ``normalize`` is set, divides by the number of + continuation tokens (length-normalised, the standard for comparing MMLU + choices of differing token length). + """ + prompt_ids = tokenizer(prompt, return_tensors="pt")["input_ids"][0] + full_ids = tokenizer(prompt + continuation, return_tensors="pt")["input_ids"][0] + cont_len = full_ids.shape[0] - prompt_ids.shape[0] + if cont_len <= 0: + return float("-inf") + + input_ids = full_ids.unsqueeze(0).to(device) + logits = model(input_ids=input_ids).logits # [1, T, V] + # logits[:, t] predicts token t+1; gather over the continuation span. + log_probs = torch.log_softmax(logits[0, :-1].float(), dim=-1) # [T-1, V] + targets = full_ids[1:].to(device) # [T-1] + cont_log_probs = log_probs[-cont_len:].gather(1, targets[-cont_len:].unsqueeze(1)).squeeze(1) + total = float(cont_log_probs.sum().item()) + return total / cont_len if normalize else total diff --git a/connito/owner_eval/metrics/gsm8k_ppl.py b/connito/owner_eval/metrics/gsm8k_ppl.py new file mode 100644 index 0000000..8cb39ca --- /dev/null +++ b/connito/owner_eval/metrics/gsm8k_ppl.py @@ -0,0 +1,36 @@ +"""GSM8K perplexity metric: exp(mean LM loss) over GSM8K test samples.""" + +from __future__ import annotations + +import math +from typing import Any + +from connito.shared.evaluate import evaluate_model +from connito.owner_eval.registry import register +from connito.owner_eval.metrics.base import BaseEvaluator, make_lm_batches, prep_tokenizer, load_hf_split + + +@register +class GSM8KPerplexity(BaseEvaluator): + name = "gsm8k_ppl" + + def evaluate(self, model: Any, tokenizer: Any, device: Any, config: Any) -> dict[str, float]: + prep_tokenizer(tokenizer) + rows = load_hf_split("openai/gsm8k", "main", "test", self.n_samples(config)) + texts = [f"{r['question']}\n{r['answer']}" for r in rows] + batches = make_lm_batches( + texts, tokenizer, + seq_len=config.eval_pipeline.eval_seq_length, + batch_size=config.eval_pipeline.eval_batch_size, + ) + metrics = evaluate_model( + step=0, + model=model, + eval_dataloader=batches, + device=device, + max_eval_batches=None, + rank=0, + ) + val_loss = float(metrics["val_loss"]) + ppl = math.exp(val_loss) if math.isfinite(val_loss) else float("inf") + return {"gsm8k_ppl": ppl, "gsm8k_ppl_val_loss": val_loss} diff --git a/connito/owner_eval/metrics/gsm8k_task.py b/connito/owner_eval/metrics/gsm8k_task.py new file mode 100644 index 0000000..25e4150 --- /dev/null +++ b/connito/owner_eval/metrics/gsm8k_task.py @@ -0,0 +1,58 @@ +"""GSM8K task score: greedy-decode each question, exact-match the final number.""" + +from __future__ import annotations + +from typing import Any + +import torch + +from connito.shared.app_logging import structlog +from connito.owner_eval.registry import register +from connito.owner_eval.metrics.base import ( + BaseEvaluator, + prep_tokenizer, + load_hf_split, + extract_final_number, + extract_gsm8k_gold, + numeric_eq, +) + +logger = structlog.get_logger(__name__) + +# Minimal zero-shot framing; nudges the model to emit a final answer we can parse. +_PROMPT_TEMPLATE = "Question: {q}\nAnswer:" + + +def _build_prompt(question: str) -> str: + return _PROMPT_TEMPLATE.format(q=question) + + +@register +class GSM8KTaskScore(BaseEvaluator): + name = "gsm8k_task" + + @torch.no_grad() + def evaluate(self, model: Any, tokenizer: Any, device: Any, config: Any) -> dict[str, float]: + prep_tokenizer(tokenizer) + rows = load_hf_split("openai/gsm8k", "main", "test", self.n_samples(config)) + max_new_tokens = config.eval_pipeline.gsm8k_max_new_tokens + + correct = 0 + n = 0 + for r in rows: + n += 1 + enc = tokenizer(_build_prompt(r["question"]), return_tensors="pt").to(device) + prompt_len = enc["input_ids"].shape[1] + gen = model.generate( + **enc, + max_new_tokens=max_new_tokens, + do_sample=False, + pad_token_id=tokenizer.pad_token_id, + ) + completion = tokenizer.decode(gen[0, prompt_len:], skip_special_tokens=True) + pred = extract_final_number(completion) + gold = extract_gsm8k_gold(r["answer"]) + correct += int(numeric_eq(pred, gold)) + + acc = correct / n if n else 0.0 + return {"gsm8k_task_acc": acc, "gsm8k_task_n": float(n)} diff --git a/connito/owner_eval/metrics/mmlu.py b/connito/owner_eval/metrics/mmlu.py new file mode 100644 index 0000000..29031f5 --- /dev/null +++ b/connito/owner_eval/metrics/mmlu.py @@ -0,0 +1,55 @@ +"""MMLU accuracy: score each of the 4 choices by log-likelihood, argmax vs gold.""" + +from __future__ import annotations + +from typing import Any + +import torch + +from connito.shared.app_logging import structlog +from connito.owner_eval.registry import register +from connito.owner_eval.metrics.base import BaseEvaluator, prep_tokenizer, load_hf_split, loglikelihood + +logger = structlog.get_logger(__name__) + +_CHOICE_LABELS = ["A", "B", "C", "D"] + + +def _format_prompt(question: str, choices: list[str]) -> str: + lines = [f"Question: {question}"] + for label, choice in zip(_CHOICE_LABELS, choices): + lines.append(f"{label}. {choice}") + lines.append("Answer:") + return "\n".join(lines) + + +@register +class MMLUAccuracy(BaseEvaluator): + name = "mmlu" + + @torch.no_grad() + def evaluate(self, model: Any, tokenizer: Any, device: Any, config: Any) -> dict[str, float]: + prep_tokenizer(tokenizer) + # Seeded shuffle so an n < full set spans all 57 subjects representatively + # (the test split is ordered by subject) and stays fixed across runs. + rows = load_hf_split( + "cais/mmlu", "all", "test", self.n_samples(config), + seed=config.eval_pipeline.sample_seed, + ) + normalize = config.eval_pipeline.mmlu_length_normalize + + correct = 0 + n = 0 + for r in rows: + choices = list(r["choices"]) + prompt = _format_prompt(r["question"], choices) + scores = [ + loglikelihood(model, tokenizer, prompt, f" {label}", device, normalize=normalize) + for label in _CHOICE_LABELS[:len(choices)] + ] + pred = int(max(range(len(scores)), key=lambda i: scores[i])) + correct += int(pred == int(r["answer"])) + n += 1 + + acc = correct / n if n else 0.0 + return {"mmlu_acc": acc, "mmlu_n": float(n)} diff --git a/connito/owner_eval/registry.py b/connito/owner_eval/registry.py new file mode 100644 index 0000000..9fd7270 --- /dev/null +++ b/connito/owner_eval/registry.py @@ -0,0 +1,64 @@ +"""Evaluator registry — the pluggability core of the owner eval pipeline. + +An evaluator is any object exposing a ``name`` and an ``evaluate(model, +tokenizer, device, config) -> dict[str, float]`` method. Returning a dict lets a +single evaluator emit several scalars (e.g. an accuracy and its sample count), +each published as its own Prometheus gauge sample by the runner. + +Add a metric by subclassing ``connito.owner_eval.metrics.base.BaseEvaluator``, +decorating it with ``@register``, importing it from +``connito/owner_eval/metrics/__init__.py`` (so the decorator runs), and adding +its ``name`` to ``config.eval_pipeline.enabled_metrics``. + +This module is intentionally dependency-light (no torch/transformers) so it can +be imported and unit-tested in isolation. +""" + +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +from connito.shared.app_logging import structlog + +logger = structlog.get_logger(__name__) + + +@runtime_checkable +class Evaluator(Protocol): + name: str + + def evaluate(self, model: Any, tokenizer: Any, device: Any, config: Any) -> dict[str, float]: + ... + + +# name -> Evaluator class. Populated by @register at import time. +REGISTRY: dict[str, type] = {} + + +def register(cls: type) -> type: + """Class decorator that registers an evaluator under its ``name``.""" + name = getattr(cls, "name", None) + if not name: + raise ValueError(f"Evaluator {cls!r} must define a non-empty class attribute 'name'") + if name in REGISTRY and REGISTRY[name] is not cls: + raise ValueError(f"Duplicate evaluator name {name!r}: {REGISTRY[name]!r} vs {cls!r}") + REGISTRY[name] = cls + return cls + + +def build_enabled(config: Any) -> list[Evaluator]: + """Instantiate the evaluators named in ``config.eval_pipeline.enabled_metrics``. + + Preserves config order; silently skips unknown names (with a warning) so a + typo in one metric name doesn't take down the whole suite. + """ + enabled = list(getattr(config.eval_pipeline, "enabled_metrics", [])) + built: list[Evaluator] = [] + for name in enabled: + cls = REGISTRY.get(name) + if cls is None: + logger.warning("unknown evaluator in enabled_metrics; skipping", metric=name, + known=sorted(REGISTRY)) + continue + built.append(cls()) + return built diff --git a/connito/owner_eval/run.py b/connito/owner_eval/run.py new file mode 100644 index 0000000..bf83c62 --- /dev/null +++ b/connito/owner_eval/run.py @@ -0,0 +1,142 @@ +"""Owner eval daemon. + +Long-running process the subnet owner runs independently of any miner/validator. +Polls the cycle API; every ``eval_interval_cycles`` cycles it downloads the +latest merged full model (latest validator HF checkpoint), runs the enabled +benchmark suite, and emits the results as Prometheus metrics. + +Launch: ``python -m connito.owner_eval.run --path owner_eval.yaml`` +One-shot (bypass the cycle gate, run once, exit — for verification/smoke): + ``python -m connito.owner_eval.run --path owner_eval.yaml --once`` + (or set ``OWNER_EVAL_FORCE_RUN=1``) + +Heavy imports (torch / bittensor / transformers / prometheus) are deferred into +the functions that use them so importing this module — e.g. to unit-test the +cycle-gate predicate — stays cheap. +""" + +from __future__ import annotations + +import argparse +import os +import time +from typing import Any + +from connito.shared.app_logging import configure_logging, structlog + +logger = structlog.get_logger(__name__) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Standalone owner eval pipeline daemon") + parser.add_argument("--path", required=True, help="Path to the OwnerEvalConfig YAML file") + parser.add_argument( + "--once", + action="store_true", + help="Bypass the cycle gate, run the suite exactly once, then exit (verification/smoke).", + ) + return parser.parse_args(argv) + + +def should_run_cycle(cycle_index: int, interval: int, last_ran_cycle: int) -> bool: + """Cycle-gate predicate, factored out for testing. + + Run when the cycle index lands on the interval boundary and we haven't + already run for this exact cycle (the daemon polls several times per cycle). + """ + if cycle_index < 0: + return False + return cycle_index % interval == 0 and cycle_index != last_ran_cycle + + +def _force_run_requested(args: argparse.Namespace) -> bool: + if getattr(args, "once", False): + return True + return str(os.environ.get("OWNER_EVAL_FORCE_RUN", "")).lower() in ("1", "true", "yes") + + +def _run_once(config: Any, expert_manager: Any, subtensor: Any, wallet: Any, + tokenizer: Any, device: Any, cycle_index: int) -> None: + from connito.owner_eval import bootstrap + from connito.owner_eval.runner import run_eval_suite + + model, checkpoint = bootstrap.load_latest_full_model( + config=config, + expert_manager=expert_manager, + subtensor=subtensor, + wallet=wallet, + device=device, + ) + try: + revision = ("base" if config.eval_pipeline.model_source == "base" + else bootstrap.model_revision_label(checkpoint)) + run_eval_suite(model, tokenizer, device, config, model_revision=revision, cycle_index=cycle_index) + finally: + bootstrap.release_model(model) + + +def main_loop(config: Any, force_run: bool = False) -> None: + from connito.shared import telemetry + from connito.shared.chain import setup_chain_worker + from connito.shared.cycle import get_phase_from_api + from connito.shared.expert_manager import ExpertManager + from connito.shared.modeling.mycelia import get_base_tokenizer + from connito.owner_eval import bootstrap + from connito.owner_eval.metrics.base import prep_tokenizer + + telemetry.TelemetryManager().start_server(port=config.eval_pipeline.telemetry_port) + # "base" canary mode needs no chain identity — skip wallet/subtensor setup so + # the publish->scrape->API path can be tested without an owner wallet. + if config.eval_pipeline.model_source == "base": + wallet, subtensor = None, None + logger.info("model_source=base: skipping chain worker (canary mode)") + else: + wallet, subtensor, _lite = setup_chain_worker(config, serve=False) + expert_manager = ExpertManager(config) + tokenizer = prep_tokenizer(get_base_tokenizer(config)) + device = bootstrap.resolve_device(config) + + N = config.eval_pipeline.eval_interval_cycles + poll_interval = config.eval_pipeline.poll_interval_sec + last_ran_cycle = -1 + + logger.info("owner eval daemon started", interval_cycles=N, poll_interval_sec=poll_interval, + telemetry_port=config.eval_pipeline.telemetry_port, device=str(device), + force_run=force_run) + + if force_run: + # One-shot: evaluate immediately against whatever the latest checkpoint is. + phase = get_phase_from_api(config) + cycle_index = phase.cycle_index if phase is not None else -1 + _run_once(config, expert_manager, subtensor, wallet, tokenizer, device, cycle_index) + return + + while True: + telemetry.set_owner_eval_heartbeat() + phase = get_phase_from_api(config) + if phase is None: + logger.warning("cycle API returned no phase; will retry") + elif should_run_cycle(phase.cycle_index, N, last_ran_cycle): + logger.info("cycle gate open; running eval suite", cycle_index=phase.cycle_index) + try: + _run_once(config, expert_manager, subtensor, wallet, tokenizer, device, phase.cycle_index) + except Exception as exc: # noqa: BLE001 — keep the daemon alive across run failures + logger.warning("owner eval run failed", cycle_index=phase.cycle_index, error=str(exc), + exc_info=True) + telemetry.inc_error("owner_eval", "run_failed") + last_ran_cycle = phase.cycle_index + time.sleep(poll_interval) + + +def main(argv: list[str] | None = None) -> int: + from connito.owner_eval import bootstrap + + configure_logging() + args = parse_args(argv) + config = bootstrap.load_config(args.path) + main_loop(config, force_run=_force_run_requested(args)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/connito/owner_eval/runner.py b/connito/owner_eval/runner.py new file mode 100644 index 0000000..6fd0b32 --- /dev/null +++ b/connito/owner_eval/runner.py @@ -0,0 +1,46 @@ +"""Run the enabled evaluator suite over a loaded model and emit Prometheus metrics.""" + +from __future__ import annotations + +from typing import Any + +from connito.shared.app_logging import structlog +from connito.shared import telemetry +from connito.owner_eval.registry import build_enabled + +# Importing the metrics package registers the built-in evaluators. +import connito.owner_eval.metrics # noqa: F401 + +logger = structlog.get_logger(__name__) + + +def run_eval_suite(model: Any, tokenizer: Any, device: Any, config: Any, + model_revision: str, cycle_index: int) -> dict[str, float]: + """Evaluate every enabled metric and publish results to Prometheus. + + One failing evaluator never aborts the rest: its exception is logged, its + status gauge set to 0, and an error counted. Returns the aggregated + ``{metric_key: value}`` dict (useful for tests / one-shot runs). + """ + results: dict[str, float] = {} + evaluators = build_enabled(config) + logger.info("running owner eval suite", count=len(evaluators), + metrics=[e.name for e in evaluators], cycle_index=cycle_index, + model_revision=model_revision) + + for ev in evaluators: + try: + with telemetry.EVAL_LATENCY_SECONDS.time(): + metric_values = ev.evaluate(model, tokenizer, device, config) + for key, value in metric_values.items(): + telemetry.set_owner_eval_metric(key, float(value)) + results[key] = float(value) + telemetry.set_owner_eval_status(ev.name, ok=True) + logger.info("evaluator complete", metric=ev.name, values=metric_values) + except Exception as exc: # noqa: BLE001 — one bad metric must not sink the suite + logger.warning("evaluator failed", metric=ev.name, error=str(exc), exc_info=True) + telemetry.set_owner_eval_status(ev.name, ok=False) + telemetry.inc_error("owner_eval", ev.name) + + telemetry.set_owner_eval_run_info(model_revision=model_revision, cycle_index=cycle_index) + return results diff --git a/connito/owner_eval/text_utils.py b/connito/owner_eval/text_utils.py new file mode 100644 index 0000000..6a545b9 --- /dev/null +++ b/connito/owner_eval/text_utils.py @@ -0,0 +1,52 @@ +"""Pure-stdlib text helpers for answer extraction. + +Kept dependency-free (no torch/transformers/datasets) so the parsing logic is +unit-testable in a minimal environment and reusable across metric evaluators. +""" + +from __future__ import annotations + +import re + +# Matches an integer or decimal, optionally signed, with thousands separators +# (commas). Examples: "42", "-3.5", "1,234", "$1,234.50" -> "1,234.50". +_NUMBER_RE = re.compile(r"-?\d[\d,]*(?:\.\d+)?") + + +def extract_final_number(text: str | None) -> float | None: + """Return the last numeric value mentioned in ``text`` as a float. + + GSM8K answers put the gold value last (after ``####``) and models trained to + "show their work" emit the final answer last too, so taking the *last* match + is the standard convention. Returns ``None`` when no number is present. + """ + if not text: + return None + matches = _NUMBER_RE.findall(text) + if not matches: + return None + cleaned = matches[-1].replace(",", "") + try: + return float(cleaned) + except ValueError: + return None + + +def extract_gsm8k_gold(answer: str) -> float | None: + """Extract the gold numeric answer from a GSM8K ``answer`` field. + + GSM8K marks the final answer with ``#### ``; fall back to the last + number in the string if the marker is absent. + """ + if answer is None: + return None + tail = answer.split("####")[-1] + return extract_final_number(tail) + + +def numeric_eq(a: float | None, b: float | None, rel_tol: float = 1e-4, abs_tol: float = 1e-6) -> bool: + """Tolerant numeric equality for comparing an extracted prediction to gold.""" + if a is None or b is None: + return False + diff = abs(a - b) + return diff <= max(rel_tol * max(abs(a), abs(b)), abs_tol) diff --git a/connito/shared/config.py b/connito/shared/config.py index 773d595..762383b 100644 --- a/connito/shared/config.py +++ b/connito/shared/config.py @@ -873,6 +873,46 @@ class EvalCfg(BaseConfig): group_a_min_weight_per_validator: float = 0.03 # > 3% from at least one validator +class EvalPipelineCfg(BaseConfig): + """Config for the standalone owner-run evaluation pipeline + (``connito.owner_eval``). The owner runs this independently of any miner or + validator to benchmark the latest merged full model every N cycles and emit + the results as Prometheus metrics.""" + + # Metric names that the runner evaluates, in order. Each must be registered + # in connito.owner_eval.registry.REGISTRY (see connito/owner_eval/metrics/). + # Where the model under test comes from. "chain" (production) fetches the + # latest merged full model a validator published to HF. "base" loads the + # pretrained base model directly — no wallet/subtensor needed — for canary / + # plumbing tests of the publish->scrape->API path. + model_source: Literal["chain", "base"] = "chain" + enabled_metrics: list[str] = ["gsm8k_ppl", "gsm8k_task", "mmlu"] + # Per-metric sample counts; metrics fall back to default_n_samples when absent. + n_samples_per_metric: dict[str, int] = { + "gsm8k_ppl": 100, + "gsm8k_task": 100, + "mmlu": 100, + } + default_n_samples: PositiveInt = 100 + # Run the suite when cycle_index % eval_interval_cycles == 0. + eval_interval_cycles: PositiveInt = 5 + poll_interval_sec: PositiveInt = 60 + telemetry_port: PositiveInt = 8400 + # gsm8k_task generation settings. + gsm8k_max_new_tokens: PositiveInt = 256 + gsm8k_few_shot: int = 0 + # mmlu length-normalises per-choice log-likelihood before argmax. + mmlu_length_normalize: bool = True + # Seed for representative subsampling (e.g. MMLU spans 57 subjects ordered + # alphabetically, so a fixed seed shuffles before selecting n to avoid an + # alphabetical-subject bias). Held constant so run-to-run deltas reflect the + # model, not a different question sample. + sample_seed: int = 0 + # Tokenisation/eval batching shared by the LM-loss metrics. + eval_batch_size: PositiveInt = 1 + eval_seq_length: PositiveInt = 2048 + + class ValidatorConfig(WorkerConfig): role: str = "validator" ckpt: ValidatorCheckpointCfg = Field(default_factory=ValidatorCheckpointCfg) @@ -930,6 +970,17 @@ class OwnerConfig(WorkerConfig): ckpt: OwnerCheckpointCfg = Field(default_factory=OwnerCheckpointCfg) +class OwnerEvalConfig(ValidatorConfig): + """Config for the owner-run eval pipeline daemon. Subclasses + ``ValidatorConfig`` so it inherits the validator checkpoint paths and the + chain-fetch wiring used to pull the latest merged full model + (``fetch_model_from_chain_validator`` resolves *validator* commits). The + owner supplies their own wallet via ``chain.*``.""" + + role: str = "owner_eval" + eval_pipeline: EvalPipelineCfg = Field(default_factory=EvalPipelineCfg) + + # --------------------------- # CLI # --------------------------- diff --git a/connito/shared/telemetry.py b/connito/shared/telemetry.py index f15d54f..0470f59 100644 --- a/connito/shared/telemetry.py +++ b/connito/shared/telemetry.py @@ -290,6 +290,35 @@ def set_validator_identity(*, hotkey: str, uid: int | None, version: str, netuid MINER_TOTAL_TRAINING_TIME_HOURS = Gauge("miner_total_training_time_hours", "Total accumulated training time (hours)") MINER_PARAM_SUM = Gauge("miner_param_sum", "Sum of expert parameter values (health check)") +# Owner standalone eval pipeline (connito.owner_eval) — benchmarks the latest +# merged full model every N cycles. One generic gauge labeled by metric name so +# new evaluators don't require new metric definitions; cardinality is bounded by +# the fixed set of registered metric keys. +OWNER_EVAL_METRIC = Gauge( + "owner_eval_metric", + "Latest owner eval-suite scalar result, labeled by metric name", + ["metric"], +) +OWNER_EVAL_STATUS = Gauge( + "owner_eval_status", + "1 if the named evaluator's last run succeeded, 0 if it raised", + ["metric"], +) +OWNER_EVAL_LAST_RUN_TS = Gauge( + "owner_eval_last_run_timestamp", + "Unix timestamp of the most recent completed owner eval run", +) +# Run identity (model revision + cycle) as an Info so labelset cardinality stays +# bounded — Info.info() atomically replaces the single labelset each run. +OWNER_EVAL_RUN_INFO = Info( + "owner_eval_run", + "Identity of the latest owner eval run (model revision + cycle index)", +) +OWNER_EVAL_HEARTBEAT_TOTAL = Counter( + "owner_eval_loop_heartbeat_total", + "Owner eval daemon poll iterations; alert on rate()->0", +) + # Histograms (Latency & Sizes) EVAL_LATENCY_SECONDS = Histogram("validator_eval_latency_seconds", "Latency of run_evaluation()") MODEL_LOAD_LATENCY_SECONDS = Histogram("validator_model_load_latency_seconds", "Latency of load_model_from_path()") @@ -406,6 +435,43 @@ def set_miner_eval_status(miner_uid: int | str, reason: EvalFailureReason | str pass +def set_owner_eval_metric(metric: str, value: float) -> None: + """Publish one owner eval scalar result. Best-effort — never raises.""" + try: + OWNER_EVAL_METRIC.labels(metric=str(metric)).set(float(value)) + except Exception: + pass + + +def set_owner_eval_status(metric: str, ok: bool) -> None: + """Mark whether the named evaluator's last run succeeded. Best-effort.""" + try: + OWNER_EVAL_STATUS.labels(metric=str(metric)).set(1.0 if ok else 0.0) + except Exception: + pass + + +def set_owner_eval_run_info(model_revision: str, cycle_index: int | str) -> None: + """Stamp the latest run's model revision + cycle and bump the run timestamp. + Best-effort — telemetry must never break the eval loop.""" + try: + OWNER_EVAL_RUN_INFO.info({ + "model_revision": str(model_revision), + "cycle_index": str(cycle_index), + }) + OWNER_EVAL_LAST_RUN_TS.set(time.time()) + except Exception: + pass + + +def set_owner_eval_heartbeat() -> None: + """Increment the daemon poll-loop heartbeat. Best-effort.""" + try: + OWNER_EVAL_HEARTBEAT_TOTAL.inc() + except Exception: + pass + + def set_miner_score_snapshot( miner_uid: int | str, *, diff --git a/connito/test/test_owner_eval_bootstrap.py b/connito/test/test_owner_eval_bootstrap.py new file mode 100644 index 0000000..251c042 --- /dev/null +++ b/connito/test/test_owner_eval_bootstrap.py @@ -0,0 +1,59 @@ +"""Tests for model-source dispatch in the owner-eval bootstrap.""" + +from types import SimpleNamespace + +import pytest + +torch = pytest.importorskip("torch") + +from connito.owner_eval import bootstrap + + +def test_base_source_uses_base_loader_not_chain(monkeypatch): + sentinel = object() + calls = {"base": 0} + + def fake_base(cfg, em, dev): + calls["base"] += 1 + return sentinel + + # If the chain path were taken it would import connito.shared.model; assert + # we never get there by failing loudly if load_base_model isn't used. + monkeypatch.setattr(bootstrap, "load_base_model", fake_base) + + cfg = SimpleNamespace(eval_pipeline=SimpleNamespace(model_source="base")) + model, ckpt = bootstrap.load_latest_full_model( + config=cfg, expert_manager=None, subtensor=None, wallet=None, + device=torch.device("cpu"), + ) + assert model is sentinel + assert ckpt is None + assert calls["base"] == 1 + + +def test_default_source_is_chain(monkeypatch): + # When model_source is absent, default to chain (production) — verified by + # the base loader NOT being called and the chain import being attempted. + monkeypatch.setattr(bootstrap, "load_base_model", + lambda *a, **k: pytest.fail("should not use base loader")) + cfg = SimpleNamespace(eval_pipeline=SimpleNamespace()) # no model_source attr + + # Stub the lazily-imported load_model so we don't pull the chain stack. + import sys + import types as _types + + class FakeModel: + def eval(self): + return self + + fake = FakeModel() + stub = _types.ModuleType("connito.shared.model") + stub.load_model = lambda **k: (fake, "CKPT") + monkeypatch.setitem(sys.modules, "connito.shared.model", stub) + + model, ckpt = bootstrap.load_latest_full_model( + config=cfg, expert_manager=None, subtensor="ST", wallet="W", + device=torch.device("cpu"), + ) + assert model is fake + assert ckpt == "CKPT" diff --git a/connito/test/test_owner_eval_metrics.py b/connito/test/test_owner_eval_metrics.py new file mode 100644 index 0000000..6c8c6dd --- /dev/null +++ b/connito/test/test_owner_eval_metrics.py @@ -0,0 +1,185 @@ +"""Evaluator behavior tests with stub models/tokenizers (no real model/datasets).""" + +from types import SimpleNamespace + +import pytest + +torch = pytest.importorskip("torch") + + +def _cfg(**overrides): + ep = SimpleNamespace( + enabled_metrics=["gsm8k_ppl", "gsm8k_task", "mmlu"], + n_samples_per_metric={}, + default_n_samples=2, + eval_seq_length=128, + eval_batch_size=1, + gsm8k_max_new_tokens=16, + mmlu_length_normalize=True, + sample_seed=0, + ) + for k, v in overrides.items(): + setattr(ep, k, v) + return SimpleNamespace(eval_pipeline=ep) + + +class _Enc(dict): + """Dict that mimics HF BatchEncoding's ``.to(device)`` (returns self).""" + def to(self, device): + return self + + +class _FakeTokenizer: + """Minimal tokenizer: maps each char to its ord; supports the call shapes used.""" + pad_token = "" + eos_token = "" + pad_token_id = 0 + + def __call__(self, text, return_tensors=None, truncation=False, max_length=None, padding=False): + if isinstance(text, str): + ids = [min(ord(c), 255) for c in text][: (max_length or 10_000)] + t = torch.tensor([ids], dtype=torch.long) + return _Enc(input_ids=t, attention_mask=torch.ones_like(t)) + # batch of strings + seqs = [[min(ord(c), 255) for c in s][: (max_length or 10_000)] for s in text] + width = max(len(s) for s in seqs) + ids, mask = [], [] + for s in seqs: + pad = width - len(s) + ids.append(s + [0] * pad) + mask.append([1] * len(s) + [0] * pad) + return _Enc(input_ids=torch.tensor(ids), attention_mask=torch.tensor(mask)) + + def decode(self, ids, skip_special_tokens=True): + return "".join(chr(int(i)) for i in ids if int(i) > 0) + + +# --------------------------------------------------------------------------- +# gsm8k_ppl +# --------------------------------------------------------------------------- +def test_gsm8k_ppl_exp_of_loss(monkeypatch): + import connito.owner_eval.metrics.gsm8k_ppl as mod + + rows = [{"question": "q1", "answer": "a1\n#### 1"}, {"question": "q2", "answer": "a2\n#### 2"}] + monkeypatch.setattr(mod, "load_hf_split", lambda *a, **k: rows) + monkeypatch.setattr(mod, "evaluate_model", lambda **k: {"val_loss": 0.0}) + + out = mod.GSM8KPerplexity().evaluate(model=object(), tokenizer=_FakeTokenizer(), + device=torch.device("cpu"), config=_cfg()) + assert out["gsm8k_ppl"] == pytest.approx(1.0) # exp(0) == 1 + assert out["gsm8k_ppl_val_loss"] == 0.0 + + +def test_gsm8k_ppl_inf_loss_gives_inf(monkeypatch): + import connito.owner_eval.metrics.gsm8k_ppl as mod + monkeypatch.setattr(mod, "load_hf_split", lambda *a, **k: [{"question": "q", "answer": "a"}]) + monkeypatch.setattr(mod, "evaluate_model", lambda **k: {"val_loss": float("inf")}) + out = mod.GSM8KPerplexity().evaluate(object(), _FakeTokenizer(), torch.device("cpu"), _cfg()) + assert out["gsm8k_ppl"] == float("inf") + + +# --------------------------------------------------------------------------- +# gsm8k_task +# --------------------------------------------------------------------------- +def test_gsm8k_task_accuracy(monkeypatch): + import connito.owner_eval.metrics.gsm8k_task as mod + + rows = [ + {"question": "qa", "answer": "work\n#### 42"}, # model will say 42 -> correct + {"question": "qb", "answer": "work\n#### 99"}, # model will say 7 -> wrong + ] + monkeypatch.setattr(mod, "load_hf_split", lambda *a, **k: rows) + + completions = iter(["The answer is 42", "The answer is 7"]) + + class GenModel: + def generate(self, input_ids=None, attention_mask=None, max_new_tokens=None, + do_sample=None, pad_token_id=None): + prompt_len = input_ids.shape[1] + text = next(completions) + cont = [min(ord(c), 255) for c in text] + row = input_ids[0].tolist() + cont + return torch.tensor([row]) + + out = mod.GSM8KTaskScore().evaluate(GenModel(), _FakeTokenizer(), torch.device("cpu"), _cfg()) + assert out["gsm8k_task_n"] == 2.0 + assert out["gsm8k_task_acc"] == pytest.approx(0.5) + + +# --------------------------------------------------------------------------- +# mmlu +# --------------------------------------------------------------------------- +def test_mmlu_argmax_over_choice_loglik(monkeypatch): + import connito.owner_eval.metrics.mmlu as mod + + rows = [ + {"question": "q1", "choices": ["w", "x", "y", "z"], "answer": 2}, + {"question": "q2", "choices": ["w", "x", "y", "z"], "answer": 0}, + ] + monkeypatch.setattr(mod, "load_hf_split", lambda *a, **k: rows) + + # Make loglikelihood deterministically prefer the gold index for row 1 and a + # wrong index for row 2, so accuracy is exactly 0.5. + prefer = {0: 2, 1: 1} + call = {"row": 0, "choice": 0} + + def fake_ll(model, tokenizer, prompt, continuation, device, normalize=True): + # continuation is " A"/" B"/" C"/" D" + idx = " ABCD".strip().index(continuation.strip()) + row = call["row"] + score = 1.0 if idx == prefer[row] else 0.0 + call["choice"] += 1 + if call["choice"] == 4: + call["choice"] = 0 + call["row"] += 1 + return score + + monkeypatch.setattr(mod, "loglikelihood", fake_ll) + out = mod.MMLUAccuracy().evaluate(object(), _FakeTokenizer(), torch.device("cpu"), _cfg()) + assert out["mmlu_n"] == 2.0 + assert out["mmlu_acc"] == pytest.approx(0.5) + + +def test_mmlu_passes_sample_seed_for_representative_sampling(monkeypatch): + import connito.owner_eval.metrics.mmlu as mod + + captured = {} + + def capture_load(path, name, split, n, seed=None): + captured.update(path=path, n=n, seed=seed) + return [] + + monkeypatch.setattr(mod, "load_hf_split", capture_load) + mod.MMLUAccuracy().evaluate(object(), _FakeTokenizer(), torch.device("cpu"), + _cfg(sample_seed=123)) + # MMLU must shuffle with the configured seed (its split is subject-ordered). + assert captured["path"] == "cais/mmlu" + assert captured["seed"] == 123 + + +# --------------------------------------------------------------------------- +# loglikelihood primitive +# --------------------------------------------------------------------------- +def test_loglikelihood_prefers_higher_logit_token(): + from connito.owner_eval.metrics.base import loglikelihood + + vocab = 256 + + class LogitModel: + """Returns logits that strongly favor whatever the actual next token is, + so the continuation gets a high (near-zero) log-likelihood.""" + def __init__(self, favored: bool): + self.favored = favored + + def __call__(self, input_ids=None): + T = input_ids.shape[1] + logits = torch.zeros(1, T, vocab) + if self.favored: + for t in range(T - 1): + logits[0, t, int(input_ids[0, t + 1])] = 50.0 + return SimpleNamespace(logits=logits) + + tok = _FakeTokenizer() + favored = loglikelihood(LogitModel(True), tok, "abc", " d", torch.device("cpu"), normalize=True) + unfavored = loglikelihood(LogitModel(False), tok, "abc", " d", torch.device("cpu"), normalize=True) + assert favored > unfavored diff --git a/connito/test/test_owner_eval_registry.py b/connito/test/test_owner_eval_registry.py new file mode 100644 index 0000000..b23f7b5 --- /dev/null +++ b/connito/test/test_owner_eval_registry.py @@ -0,0 +1,84 @@ +"""Registry + cycle-gate tests (light deps: structlog only).""" + +from types import SimpleNamespace + +import pytest + +from connito.owner_eval import registry +from connito.owner_eval.registry import register, build_enabled, REGISTRY +from connito.owner_eval.run import should_run_cycle + + +@pytest.fixture +def clean_registry(): + saved = dict(REGISTRY) + REGISTRY.clear() + try: + yield REGISTRY + finally: + REGISTRY.clear() + REGISTRY.update(saved) + + +def _cfg(metrics): + return SimpleNamespace(eval_pipeline=SimpleNamespace(enabled_metrics=metrics)) + + +def test_register_populates_registry(clean_registry): + @register + class Foo: + name = "foo" + def evaluate(self, *a): # pragma: no cover - not called + return {} + + assert clean_registry["foo"] is Foo + + +def test_register_requires_name(clean_registry): + with pytest.raises(ValueError): + @register + class NoName: + name = "" + + +def test_register_rejects_duplicate(clean_registry): + @register + class A: + name = "dup" + with pytest.raises(ValueError): + @register + class B: + name = "dup" + + +def test_build_enabled_order_and_skip_unknown(clean_registry): + @register + class M1: + name = "m1" + @register + class M2: + name = "m2" + + built = build_enabled(_cfg(["m2", "unknown", "m1"])) + # preserves config order, drops unknown + assert [type(e).__name__ for e in built] == ["M2", "M1"] + + +def test_builtin_metrics_register_on_import(): + # Importing the metrics package must register the three built-ins. + pytest.importorskip("torch") + import connito.owner_eval.metrics # noqa: F401 + for name in ("gsm8k_ppl", "gsm8k_task", "mmlu"): + assert name in REGISTRY + + +@pytest.mark.parametrize("cycle,interval,last,expected", [ + (5, 5, -1, True), + (5, 5, 5, False), # already ran this cycle + (4, 5, -1, False), + (10, 5, 5, True), + (0, 5, -1, True), # cycle 0 is on the boundary + (-1, 5, -1, False), # no phase / unknown +]) +def test_should_run_cycle(cycle, interval, last, expected): + assert should_run_cycle(cycle, interval, last) is expected diff --git a/connito/test/test_owner_eval_runner.py b/connito/test/test_owner_eval_runner.py new file mode 100644 index 0000000..22762be --- /dev/null +++ b/connito/test/test_owner_eval_runner.py @@ -0,0 +1,53 @@ +"""Runner tests: gauge emission + failure isolation (needs prometheus_client).""" + +from types import SimpleNamespace + +import pytest + +pytest.importorskip("prometheus_client") +pytest.importorskip("torch") + + +def _cfg(metrics): + return SimpleNamespace(eval_pipeline=SimpleNamespace(enabled_metrics=metrics)) + + +def test_run_eval_suite_emits_gauges_and_isolates_failures(monkeypatch): + from connito.owner_eval import runner + from connito.owner_eval.registry import REGISTRY + from connito.shared import telemetry + + saved = dict(REGISTRY) + REGISTRY.clear() + try: + class Good: + name = "good" + def evaluate(self, model, tokenizer, device, config): + return {"good_x": 1.0, "good_y": 2.0} + + class Bad: + name = "bad" + def evaluate(self, model, tokenizer, device, config): + raise RuntimeError("boom") + + REGISTRY["good"] = Good + REGISTRY["bad"] = Bad + + results = runner.run_eval_suite( + model=object(), tokenizer=object(), device="cpu", + config=_cfg(["good", "bad"]), + model_revision="globalver_7", cycle_index=5, + ) + + # good metrics returned and published + assert results == {"good_x": 1.0, "good_y": 2.0} + assert telemetry.OWNER_EVAL_METRIC.labels(metric="good_x")._value.get() == 1.0 + assert telemetry.OWNER_EVAL_METRIC.labels(metric="good_y")._value.get() == 2.0 + # status gauges reflect success/failure independently + assert telemetry.OWNER_EVAL_STATUS.labels(metric="good")._value.get() == 1.0 + assert telemetry.OWNER_EVAL_STATUS.labels(metric="bad")._value.get() == 0.0 + # run info timestamp was stamped + assert telemetry.OWNER_EVAL_LAST_RUN_TS._value.get() > 0 + finally: + REGISTRY.clear() + REGISTRY.update(saved) diff --git a/connito/test/test_owner_eval_text_utils.py b/connito/test/test_owner_eval_text_utils.py new file mode 100644 index 0000000..efb551b --- /dev/null +++ b/connito/test/test_owner_eval_text_utils.py @@ -0,0 +1,47 @@ +"""Pure-logic tests for owner-eval answer extraction (no heavy deps).""" + +import pytest + +from connito.owner_eval.text_utils import ( + extract_final_number, + extract_gsm8k_gold, + numeric_eq, +) + + +@pytest.mark.parametrize("text,expected", [ + ("The answer is 42.", 42.0), + ("So we get 1,234 apples", 1234.0), + ("It costs $1,234.50 total", 1234.50), + ("first 3 then 7 finally 9", 9.0), + ("negative result: -15", -15.0), + ("3.14159", 3.14159), + ("no numbers here", None), + ("", None), + (None, None), +]) +def test_extract_final_number(text, expected): + assert extract_final_number(text) == expected + + +def test_extract_gsm8k_gold_uses_marker(): + answer = "Janet has 16 eggs ... she makes $18 per day.\n#### 18" + assert extract_gsm8k_gold(answer) == 18.0 + + +def test_extract_gsm8k_gold_falls_back_without_marker(): + assert extract_gsm8k_gold("the result is 7") == 7.0 + assert extract_gsm8k_gold(None) is None + + +@pytest.mark.parametrize("a,b,expected", [ + (42.0, 42.0, True), + (42.0, 42.00001, True), # within tolerance + (42.0, 43.0, False), + (None, 42.0, False), + (42.0, None, False), + (None, None, False), + (0.0, 0.0, True), +]) +def test_numeric_eq(a, b, expected): + assert numeric_eq(a, b) is expected diff --git a/observability/prometheus.yml b/observability/prometheus.yml index 9a6eeed..0f8c8b9 100644 --- a/observability/prometheus.yml +++ b/observability/prometheus.yml @@ -12,3 +12,10 @@ scrape_configs: static_configs: # Expecting miner 0 on 8100, miner 1 on 8101, etc. - targets: ['localhost:8100'] + + - job_name: 'mycelia_owner_eval' + static_configs: + # Standalone owner eval daemon (connito.owner_eval). Publishes owner_eval_* + # metrics on eval_pipeline.telemetry_port (default 8400). Point this at the + # host:port where the owner runs the daemon. + - targets: ['localhost:8400'] diff --git a/owner_eval.canary.yaml b/owner_eval.canary.yaml new file mode 100644 index 0000000..7a20ab1 --- /dev/null +++ b/owner_eval.canary.yaml @@ -0,0 +1,31 @@ +# Canary config for the owner eval daemon — validates the publish -> scrape -> +# API path end to end with minimal cost. NOT for production. +# +# Quick one-shot (fires immediately, no cycle wait, no wallet): +# CUDA_VISIBLE_DEVICES=5 .venv/bin/python -m connito.owner_eval.run \ +# --path owner_eval.canary.yaml --once +# +# Continuous, gated on the live cycle API (fires every cycle): +# CUDA_VISIBLE_DEVICES=5 .venv/bin/python -m connito.owner_eval.run \ +# --path owner_eval.canary.yaml +# +# Then: curl localhost:8400/metrics | grep owner_eval +# and confirm the 'mycelia_owner_eval' Prometheus target is UP. + +model: + device: cuda + precision: bf16-mixed + +eval_pipeline: + model_source: base # canary: pretrained base model, no chain/wallet + enabled_metrics: ["gsm8k_ppl", "gsm8k_task", "mmlu"] + n_samples_per_metric: + gsm8k_ppl: 2 + gsm8k_task: 2 + mmlu: 2 + eval_interval_cycles: 1 # fire every cycle (continuous mode) + poll_interval_sec: 30 + telemetry_port: 8400 + gsm8k_max_new_tokens: 32 + eval_seq_length: 512 + sample_seed: 0 diff --git a/pyproject.toml b/pyproject.toml index fb673aa..90bc126 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,6 +32,7 @@ Issues = "https://github.com/CrucibleAILabs/subnet-MoE/issues" [project.scripts] weightnet-miner = "connito.miner.cli:main" weightnet-validator = "connito.validator.cli:main" +weightnet-owner-eval = "connito.owner_eval.cli:main" # Hatch build config — package only our src tree [tool.hatch.build.targets.wheel]