serving: release_id keys audits, quarantine and the READY pool; gateway routes by model

docker/attest/attest_server.py

@@ -40,6 +40,7 @@ def run(args, timeout=120.0):
         }
     out = json.loads(proc.stdout)
     out['queued_ms'] = round(queued_ms, 1)
+    out['version'] = VERSION
     return 200, out
 
 

gittensor/serving/api.py

@@ -33,7 +33,7 @@
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from gittensor.constants import SERVING_MAX_TOKENS
-from gittensor.serving.loadout import ServingRelease
+from gittensor.serving.loadout import ServingLoadout, ServingRelease
 from gittensor.serving.state import ReadyMiner, RequestRecord, ServedRequest, ServingState, finite_or_none
 from gittensor.serving.stream import SSE_DONE, Event, consume_stream, sse_event
 from gittensor.synapses import InferenceSynapse
@@ -50,13 +50,14 @@ def parse_api_keys(raw: Optional[str]) -> Set[str]:
 
 def build_app(
     state: ServingState,
-    release: ServingRelease,
+    loadout: ServingLoadout,
     api_keys: Set[str],
     dendrite_factory,
     request_timeout: float,
     baseline_keys: Optional[Set[str]] = None,
 ) -> FastAPI:
     baseline = set(baseline_keys or ())
+    primary = loadout.primary
     app = FastAPI(title='Gittensor Serving API', version='0.1.0-beta')
     dendrite_holder: Dict[str, bt.Dendrite] = {}
 
@@ -87,20 +88,24 @@ async def models(_: str = Depends(require_key)):
             'object': 'list',
             'data': [
                 {
-                    'id': release.model_id,
+                    'id': release.release_id,
                     'object': 'model',
                     'owned_by': 'gittensor',
                     # release identity (contract P2): what every READY miner behind this endpoint is verified against
+                    'model_id': release.model_id,
                     'runtime_pin': release.runtime_pin,
                     'model_sha256': release.model_sha256,
                 }
+                for release in loadout.releases
             ],
         }
 
     @app.get('/v1/serving/status')
     async def status(_: str = Depends(require_key)):
         snap = state.snapshot()
-        snap['model_id'] = release.model_id
+        snap['model_id'] = primary.model_id
+        snap['release_id'] = primary.release_id
+        snap['releases'] = [r.release_id for r in loadout.releases]
         snap['recent'] = [r.__dict__ for r in state.recent(50)]
         return snap
 
@@ -119,6 +124,11 @@ async def chat_completions(request: Request, key: str = Depends(require_key)):
             )
         if body.get('n', 1) != 1:
             raise HTTPException(status_code=400, detail='n must be 1')
+        wanted = body.get('model')
+        try:  # `model` = a release_id (or a model_id: its first release); absent -> the primary release
+            release = loadout.get(str(wanted)) if wanted else primary
+        except KeyError:
+            raise HTTPException(status_code=404, detail=f'model {wanted!r} is not served; see /v1/models')
         try:
             max_tokens = int(body.get('max_tokens') or body.get('max_completion_tokens') or release.max_tokens)
         except (TypeError, ValueError):
@@ -127,7 +137,7 @@ async def chat_completions(request: Request, key: str = Depends(require_key)):
         want_logprobs = bool(body.get('logprobs', False))
         want_stream = bool(body.get('stream', False))
 
-        miner = state.acquire(release.model_id, probation=key in baseline)
+        miner = state.acquire(release.release_id, probation=key in baseline)
         if miner is None:
             raise HTTPException(status_code=429, detail='no READY serving capacity')
         inflight = state.inflight().get(miner.uid, 1)
@@ -146,6 +156,7 @@ def finish(result: Optional[InferenceSynapse]) -> bool:
                     uid=miner.uid,
                     hotkey=miner.hotkey,
                     model_id=release.model_id,
+                    release_id=release.release_id,
                     messages=messages,
                     ok=ok and (result.served_model_id == release.model_id if result else False),
                     latency_ms=latency_ms,
@@ -265,7 +276,13 @@ async def _dispatch(
     on_event: Optional[Callable[[Event], Awaitable[None]]] = None,
 ) -> InferenceSynapse:
     # logprobs always requested so organic traffic is indistinguishable from audits on the wire.
-    synapse = InferenceSynapse(messages=messages, model_id=release.model_id, max_tokens=max_tokens, logprobs=True)
+    synapse = InferenceSynapse(
+        messages=messages,
+        model_id=release.model_id,
+        release_id=release.release_id,
+        max_tokens=max_tokens,
+        logprobs=True,
+    )
     return await consume_stream(dendrite, miner.axon, synapse, timeout, on_event)
 
 
@@ -295,7 +312,7 @@ def stop(self):
 
 def start_serving_api(
     state: ServingState,
-    release: ServingRelease,
+    loadout: ServingLoadout,
     wallet: bt.Wallet,
     api_keys: Set[str],
     host: str,
@@ -307,7 +324,7 @@ def start_serving_api(
         raise ValueError('SERVING_API_KEYS is empty; refusing to start without API keys')
     app = build_app(
         state,
-        release,
+        loadout,
         api_keys | set(baseline_keys or ()),
         lambda: bt.Dendrite(wallet=wallet),
         request_timeout,

gittensor/serving/audit.py

@@ -132,6 +132,7 @@ class WindowVerdict:
     mean: float
     threshold: float
     quarantined_until: float = 0.0
+    strikes: int = 0  # lifetime wrong answers on this (hotkey, release)
 
     def as_dict(self) -> dict:
         return {
@@ -140,6 +141,7 @@ def as_dict(self) -> dict:
             'mean': round(self.mean, 4),
             'threshold': round(self.threshold, 4),
             'quarantined_until': round(self.quarantined_until, 1),
+            'strikes': self.strikes,
         }
 
 
@@ -154,51 +156,60 @@ class AuditWindow:
     thresholds: Sequence[Tuple[int, float]] = SERVING_AUDIT_WINDOW_THRESHOLDS
     quarantine_s: float = SERVING_QUARANTINE_S
     _values: Dict[Tuple[str, str], Deque[float]] = field(default_factory=dict)
-    _quarantine: Dict[Tuple[str, str], float] = field(default_factory=dict)  # (hotkey, model) -> until ts
+    _quarantine: Dict[Tuple[str, str], float] = field(default_factory=dict)  # (hotkey, release) -> until ts
+    _strikes: Dict[Tuple[str, str], int] = field(default_factory=dict)  # (hotkey, release) -> wrong answers, ever
 
-    def record(self, hotkey: str, model_id: str, value: float) -> None:
-        key = (hotkey, model_id)
+    def record(self, hotkey: str, release_id: str, value: float) -> None:
+        key = (hotkey, release_id)
         if key not in self._values:
             self._values[key] = deque(maxlen=self.size)
         self._values[key].append(max(0.0, min(1.0, float(value))))
 
-    def strike(self, hotkey: str, model_id: str, now: Optional[float] = None) -> float:
+    def strike(self, hotkey: str, release_id: str, now: Optional[float] = None) -> float:
         """A wrong answer: wipe the window and quarantine the (hotkey, release) until the returned timestamp."""
-        key = (hotkey, model_id)
+        key = (hotkey, release_id)
         self._values.pop(key, None)
+        self._strikes[key] = self._strikes.get(key, 0) + 1
         until = (now if now is not None else time.time()) + self.quarantine_s
         self._quarantine[key] = until
         return until
 
-    def quarantined_until(self, hotkey: str, model_id: str, now: Optional[float] = None) -> float:
-        until = self._quarantine.get((hotkey, model_id), 0.0)
+    def strikes(self, hotkey: str, release_id: str) -> int:
+        return self._strikes.get((hotkey, release_id), 0)
+
+    def quarantined_until(self, hotkey: str, release_id: str, now: Optional[float] = None) -> float:
+        until = self._quarantine.get((hotkey, release_id), 0.0)
         return until if until > (now if now is not None else time.time()) else 0.0
 
     def to_dict(self) -> dict:
         return {
             'size': self.size,
-            'values': [[hk, mid, list(xs)] for (hk, mid), xs in self._values.items()],
-            'quarantine': [[hk, mid, until] for (hk, mid), until in self._quarantine.items()],
+            'values': [[hk, rid, list(xs)] for (hk, rid), xs in self._values.items()],
+            'quarantine': [[hk, rid, until] for (hk, rid), until in self._quarantine.items()],
+            'strikes': [[hk, rid, n] for (hk, rid), n in self._strikes.items()],
         }
 
     @classmethod
     def from_dict(cls, raw: dict, **kwargs) -> 'AuditWindow':
         window = cls(**kwargs)
-        for hk, mid, xs in raw.get('values', []):
+        for hk, rid, xs in raw.get('values', []):
             for x in xs[-window.size :]:
-                window.record(str(hk), str(mid), float(x))
-        for hk, mid, until in raw.get('quarantine', []):
-            window._quarantine[(str(hk), str(mid))] = float(until)
+                window.record(str(hk), str(rid), float(x))
+        for hk, rid, until in raw.get('quarantine', []):
+            window._quarantine[(str(hk), str(rid))] = float(until)
+        for hk, rid, n in raw.get('strikes', []):
+            window._strikes[(str(hk), str(rid))] = int(n)
         return window
 
-    def verdict(self, hotkey: str, model_id: str, now: Optional[float] = None) -> WindowVerdict:
-        until = self.quarantined_until(hotkey, model_id, now)
-        xs = self._values.get((hotkey, model_id))
+    def verdict(self, hotkey: str, release_id: str, now: Optional[float] = None) -> WindowVerdict:
+        until = self.quarantined_until(hotkey, release_id, now)
+        strikes = self.strikes(hotkey, release_id)
+        xs = self._values.get((hotkey, release_id))
         if not xs:
-            return WindowVerdict(False, 0, 0.0, float('inf'), until)
+            return WindowVerdict(False, 0, 0.0, float('inf'), until, strikes)
         mean = sum(xs) / len(xs)
         threshold = window_threshold(len(xs), self.thresholds)
-        return WindowVerdict(mean >= threshold and until == 0.0, len(xs), mean, threshold, until)
+        return WindowVerdict(mean >= threshold and until == 0.0, len(xs), mean, threshold, until, strikes)
 
 
 class Reference(Protocol):
@@ -383,15 +394,16 @@ def verify_served(
     token_ids: Optional[Sequence[int]] = None,
     end_of_turn: Sequence[str] = ('<|im_end|>', '<|endoftext|>', '</s>'),
     token_bytes: Optional[Sequence[Sequence[int]]] = None,
+    release: Optional[ServingRelease] = None,
 ) -> AuditVerdict:
     """Verify a served (greedy) completion by teacher forcing it under the reference.
 
     The reference's argmax at each position is what an honest copy would have generated, so the miner's tokens
     must match it and the miner's logprobs must match the reference's logprob of that same token. When the miner
     reported ``token_ids`` the reference forces exactly that sequence; otherwise it re-tokenizes the text, and a
     text that re-tokenizes to a different length (a greedy decode is not always the canonical tokenization of
-    its own output) is not comparable position by position and counts as a soft miss. The bands failing on
-    aligned lengths is a wrong answer (``hard``). A reference that echoes the forced tokens' bytes lets the ids
+    its own output) is not comparable position by position and counts as a soft miss. The bands (the release's,
+    else the constants) failing on aligned lengths is a wrong answer (``hard``). A reference that echoes the forced tokens' bytes lets the ids
     be bound to the text the user actually received.
     """
     if completion is None or not tokens or token_logprobs is None or len(tokens) != len(token_logprobs):
@@ -422,4 +434,13 @@ def verify_served(
             if '\ufffd' not in text and '\ufffd' not in completion and text != completion:
                 return AuditVerdict(False, 0.0, float('inf'), 'token ids do not spell the completion')
     case = AuditCase(messages=list(messages), max_tokens=len(mine), reference_tokens=argmax, reference_logprobs=ref_lp)
-    return verify_response(case, mine, mine_lp)
+    if release is None:
+        return verify_response(case, mine, mine_lp)
+    return verify_response(
+        case,
+        mine,
+        mine_lp,
+        release.min_prefix_agreement,
+        release.max_mean_abs_logprob_diff,
+        release.max_abs_logprob_diff,
+    )

gittensor/serving/loadout.py

@@ -28,6 +28,12 @@
 
 import bittensor as bt
 
+from gittensor.constants import (
+    SERVING_AUDIT_MAX_ABS_LOGPROB_DIFF,
+    SERVING_AUDIT_MAX_MEAN_ABS_LOGPROB_DIFF,
+    SERVING_AUDIT_MIN_PREFIX_AGREEMENT,
+)
+
 WEIGHTS_DIR = Path(__file__).parent.parent / 'validator' / 'weights'
 DEFAULT_LOADOUT_PATH = WEIGHTS_DIR / 'serving_loadout.json'
 ECHO_LOADOUT_PATH = WEIGHTS_DIR / 'serving_loadout.echo.json'
@@ -37,6 +43,7 @@
 class ServingRelease:
     model_id: str
     backend: str
+    release_id: str = ''  # what the validator keys audits, quarantine and READY sets by; defaults to model_id
     max_tokens: int = 64
     base_url: Optional[str] = None  # miner side: the runtime this miner serves from
     runtime_pin: Optional[str] = None
@@ -63,14 +70,43 @@ class ServingRelease:
     vram_model_reserved_bytes: Optional[float] = None
     ttft_full_ms: Optional[float] = None  # validator-observed TTFT up to which latency credit is 1.0
     ttft_zero_ms: Optional[float] = None  # ... and at which it reaches 0.0
+    # Audit bands for this release (``audit`` block); None -> the constants' defaults, which are calibrated for a
+    # bit-reproducible pin. A runtime that is not deterministic ships its own bands here.
+    audit_min_prefix_agreement: Optional[float] = None
+    audit_max_mean_abs_logprob_diff: Optional[float] = None
+    audit_max_abs_logprob_diff: Optional[float] = None
+
+    def __post_init__(self) -> None:
+        if not self.release_id:
+            self.release_id = self.model_id
+
+    @property
+    def min_prefix_agreement(self) -> float:
+        if self.audit_min_prefix_agreement is None:
+            return SERVING_AUDIT_MIN_PREFIX_AGREEMENT
+        return self.audit_min_prefix_agreement
+
+    @property
+    def max_mean_abs_logprob_diff(self) -> float:
+        if self.audit_max_mean_abs_logprob_diff is None:
+            return SERVING_AUDIT_MAX_MEAN_ABS_LOGPROB_DIFF
+        return self.audit_max_mean_abs_logprob_diff
+
+    @property
+    def max_abs_logprob_diff(self) -> float:
+        if self.audit_max_abs_logprob_diff is None:
+            return SERVING_AUDIT_MAX_ABS_LOGPROB_DIFF
+        return self.audit_max_abs_logprob_diff
 
     @classmethod
     def from_dict(cls, raw: dict) -> 'ServingRelease':
         speed = raw.get('speed') or {}
         attest = raw.get('attest') or {}
+        audit = raw.get('audit') or {}
         return cls(
             model_id=raw['model_id'],
             backend=raw['backend'],
+            release_id=str(raw.get('release_id') or raw['model_id']),
             max_tokens=int(raw.get('max_tokens', 64)),
             base_url=raw.get('base_url'),
             runtime_pin=raw.get('runtime_pin'),
@@ -89,6 +125,9 @@ def from_dict(cls, raw: dict) -> 'ServingRelease':
             vram_model_reserved_bytes=_optional_float(attest.get('vram_model_reserved_bytes')),
             ttft_full_ms=_optional_float(speed.get('ttft_full_ms')),
             ttft_zero_ms=_optional_float(speed.get('ttft_zero_ms')),
+            audit_min_prefix_agreement=_optional_float(audit.get('min_prefix_agreement')),
+            audit_max_mean_abs_logprob_diff=_optional_float(audit.get('max_mean_abs_logprob_diff')),
+            audit_max_abs_logprob_diff=_optional_float(audit.get('max_abs_logprob_diff')),
         )
 
 
@@ -117,20 +156,24 @@ class ServingLoadout:
     def __post_init__(self) -> None:
         if not self.releases:
             raise ValueError('serving loadout has no releases')
-        ids = [r.model_id for r in self.releases]
+        ids = [r.release_id for r in self.releases]
         if len(set(ids)) != len(ids):
-            raise ValueError(f'duplicate model_id in serving loadout: {ids}')
+            raise ValueError(f'duplicate release_id in serving loadout: {ids}')
 
     @property
     def primary(self) -> ServingRelease:
         """The first release: what the inference API serves and what a miner runs unless SERVING_RELEASE says otherwise."""
         return self.releases[0]
 
-    def get(self, model_id: str) -> ServingRelease:
+    def get(self, release_id: str) -> ServingRelease:
+        """By release_id, then by model_id (the first release serving that model)."""
+        for release in self.releases:
+            if release.release_id == release_id:
+                return release
         for release in self.releases:
-            if release.model_id == model_id:
+            if release.model_id == release_id:
                 return release
-        raise KeyError(f'release {model_id!r} not in serving loadout: {[r.model_id for r in self.releases]}')
+        raise KeyError(f'release {release_id!r} not in serving loadout: {[r.release_id for r in self.releases]}')
 
 
 def resolve_loadout_path(path: Optional[Path] = None) -> Path:
@@ -161,7 +204,7 @@ def load_serving_loadout(path: Optional[Path] = None) -> ServingLoadout:
         loadout.primary.reference_api_key = key_override
 
     lines = [
-        f'Serving release: model={release.model_id} backend={release.backend} pin={release.runtime_pin} '
+        f'Serving release {release.release_id}: model={release.model_id} backend={release.backend} pin={release.runtime_pin} '
         f'reference={"live " + release.reference_url if release.reference_url else (release.audit_bank or "echo")}'
         for release in loadout.releases
     ]