validator: a rubric grader for text tasks, and stop and token limits taken from the arena

microtensor_subnet/validator/context.py

@@ -18,6 +18,8 @@
 SYSTEM = "system"
 ADVERSARIAL = "adversarial"
 TESTS = (SLM, HARNESS, ROUTER, ESCALATION, SYSTEM)
+SQL_STOP = ("\n\n", ";")
+SQL_MAX_TOKENS = 256
 
 
 @dataclass(slots=True)
@@ -106,6 +108,18 @@ class EpochContext:
     def grader(self) -> str:
         return str(self.definition.get("grader") or "sqlite_execution")
 
+    @property
+    def stop(self) -> tuple[str, ...]:
+        block = self.definition.get("generation") or {}
+        found = block.get("stop", SQL_STOP) if isinstance(block, Mapping) else SQL_STOP
+        return tuple(str(s) for s in found)
+
+    @property
+    def max_tokens(self) -> int:
+        block = self.definition.get("generation") or {}
+        found = block.get("max_tokens") if isinstance(block, Mapping) else None
+        return int(found or SQL_MAX_TOKENS)
+
     @property
     def slm_prompt(self) -> str:
         return str(

microtensor_subnet/validator/grade.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import json
+import re
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, Final
@@ -8,6 +10,9 @@
 
 SQL_EXECUTION: Final[str] = "sqlite_execution"
 EXACT_TEXT: Final[str] = "exact_text"
+RUBRIC_TEXT: Final[str] = "rubric_text"
+WORD = re.compile(r"[A-Za-z][A-Za-z0-9'\-]*")
+FENCE = re.compile(r"^```[A-Za-z]*\s*|\s*```$")
 
 
 class GradeError(ValueError):
@@ -60,13 +65,94 @@ def grade_exact(output: Any, label: Mapping[str, Any]) -> Grade:
     return Grade(1.0, True) if expected and actual == expected else Grade(0.0, False, "mismatch")
 
 
+def _parsed(output: Any) -> Any:
+    if isinstance(output, Mapping):
+        return output
+    text = FENCE.sub("", str(output if output is not None else "").strip())
+    try:
+        return json.loads(text)
+    except ValueError:
+        return text
+
+
+def _same(actual: Any, expected: Any) -> bool:
+    if isinstance(expected, str):
+        return isinstance(actual, str) and actual.strip() == expected.strip()
+    return bool(actual == expected)
+
+
+def _sentence_case(text: str, proper: Sequence[str]) -> bool:
+    words = WORD.findall(text)
+    if not words or not words[0][0].isupper():
+        return False
+    allowed = {str(p).lower() for p in proper}
+    for word in words[1:]:
+        titled = len(word) > 1 and word[0].isupper() and word[1:].islower()
+        if titled and word.lower() not in allowed:
+            return False
+    return True
+
+
+def _bounded(value: int, bounds: Any) -> bool:
+    low, high = [*list(bounds), None, None][:2]
+    return (low is None or value >= int(low)) and (high is None or value <= int(high))
+
+
+def grade_rubric(output: Any, label: Mapping[str, Any]) -> Grade:
+    rules = label.get("rubric")
+    if not isinstance(rules, Mapping) or not rules:
+        raise GradeError("the item label carries no rubric")
+    parsed = _parsed(output)
+    wants_json = rules.get("format") == "json"
+    if wants_json and not isinstance(parsed, Mapping):
+        return Grade(0.0, False, "not a JSON object")
+    field = str(rules.get("field") or "")
+    if isinstance(parsed, Mapping):
+        text = str(parsed.get(field, "")) if field else json.dumps(parsed, sort_keys=True)
+    else:
+        text = str(parsed)
+    checks: list[tuple[bool, str]] = []
+    if field and isinstance(parsed, Mapping):
+        checks.append((isinstance(parsed.get(field), str), f"{field} is missing"))
+    for key, expected in (rules.get("equals") or {}).items():
+        actual = parsed.get(key) if isinstance(parsed, Mapping) else None
+        checks.append((_same(actual, expected), f"{key} differs"))
+    for key, expected in (rules.get("set_equals") or {}).items():
+        actual = parsed.get(key) if isinstance(parsed, Mapping) else None
+        same = isinstance(actual, list) and {str(a).strip() for a in actual} == {
+            str(e).strip() for e in expected
+        }
+        checks.append((same, f"{key} differs"))
+    if "text_equals" in rules:
+        checks.append((text.strip() == str(rules["text_equals"]).strip(), "text differs"))
+    if "words" in rules:
+        checks.append((_bounded(len(text.split()), rules["words"]), "word count out of bounds"))
+    if "chars" in rules:
+        checks.append((_bounded(len(text.strip()), rules["chars"]), "length out of bounds"))
+    lowered = text.lower()
+    if rules.get("include_any"):
+        hit = any(str(k).lower() in lowered for k in rules["include_any"])
+        checks.append((hit, "misses every key term"))
+    for banned in rules.get("exclude") or []:
+        checks.append((str(banned).lower() not in lowered, f"contains {banned!r}"))
+    if rules.get("case") == "sentence":
+        checks.append((_sentence_case(text, rules.get("proper") or []), "not sentence case"))
+    if not checks:
+        raise GradeError("the rubric names no checks")
+    failed = [reason for ok, reason in checks if not ok]
+    metric = (len(checks) - len(failed)) / len(checks)
+    return Grade(metric, not failed, failed[0] if failed else "")
+
+
 def grade(
     grader: str, output: Any, label: Mapping[str, Any], databases: Mapping[str, bytes]
 ) -> Grade:
     if grader == SQL_EXECUTION:
         return grade_sql(output, label, databases)
     if grader == EXACT_TEXT:
         return grade_exact(output, label)
+    if grader == RUBRIC_TEXT:
+        return grade_rubric(output, label)
     raise GradeError(f"unknown grader {grader!r}")
 
 

microtensor_subnet/validator/pools/harness.py

@@ -75,14 +75,19 @@ def run_condition(
 
 
 def run_bare(
-    context: EpochContext, client: SlmClient, items: Sequence[Item], *, max_tokens: int = 256
+    context: EpochContext,
+    client: SlmClient,
+    items: Sequence[Item],
+    *,
+    max_tokens: int | None = None,
 ) -> Condition:
+    budget = max_tokens or context.max_tokens
     runs: list[ItemRun] = []
     tokens = 0
     for item in items:
         prompt = render_prompt(context.slm_prompt, item.task)
         try:
-            generation = client.generate(prompt, max_tokens=max_tokens, stop=("\n\n", ";"))
+            generation = client.generate(prompt, max_tokens=budget, stop=context.stop)
         except SlmError as exc:
             runs.append(_graded(context, item, "", 0.0, False, str(exc)))
             continue

microtensor_subnet/validator/pools/slm.py

@@ -11,13 +11,12 @@
 from microtensor_subnet.validator.sandbox.slm import SlmClient, SlmError, gpu_memory_used
 
 log = logging.getLogger("microtensor_subnet.validator.pools.slm")
-STOP = ("\n\n", ";")
 
 
 def _run_item(context: EpochContext, client: SlmClient, item: Item, max_tokens: int) -> ItemRun:
     prompt = render_prompt(context.slm_prompt, item.task)
     try:
-        generation = client.generate(prompt, max_tokens=max_tokens, stop=STOP)
+        generation = client.generate(prompt, max_tokens=max_tokens, stop=context.stop)
     except SlmError as exc:
         return ItemRun(item.item_id, item.family, "", False, 0.0, str(exc), answered=False)
     try:
@@ -46,19 +45,20 @@ def run(
     *,
     gpu: str,
     parameters: int,
-    max_tokens: int = 256,
+    max_tokens: int | None = None,
     items: Sequence[Item] | None = None,
 ) -> SlmRun:
+    budget = max_tokens or context.max_tokens
     chosen = list(items) if items is not None else context.items.originals(SLM)
     twins = context.items.twins_of(SLM)
     runs: list[ItemRun] = []
     twin_runs: list[ItemRun] = []
     peak = gpu_memory_used(gpu)
     for item in chosen:
-        runs.append(_run_item(context, client, item, max_tokens))
+        runs.append(_run_item(context, client, item, budget))
         twin = twins.get(item.item_id)
         if twin is not None:
-            twin_runs.append(_run_item(context, client, twin, max_tokens))
+            twin_runs.append(_run_item(context, client, twin, budget))
         peak = max(peak, gpu_memory_used(gpu))
     answered = [r for r in runs if r.answered]
     return SlmRun(