File size: 2,324 Bytes
658bb70 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """
Hugging Face Inference Endpoints custom handler.
Deliberately narrow: this endpoint scores an ALREADY-EXTRACTED feature vector — it
does not accept or walk a raw serialization graph. Graph feature extraction stays
single-sourced in the TS engine (packages/intelligence/src/features.ts); see
docs/huggingface-model-plan.md §0 for why duplicating that traversal here would be
an unacceptable drift risk for a system whose whole premise is "never fabricate."
Expected input (feature order MUST match model.json's `featureNames`):
{"inputs": {"features": [0.12, 3, 1, 2, 1, 1, 5, 0.0]}}
or a batch:
{"inputs": [{"features": [...]}, {"features": [...]}]}
Output mirrors the TS AnomalyOutput contract exactly (see scorer.py / model.ts):
{"score": 0.87, "uncertainty": 0.05, "explanation": "learned anomaly 87% (top: route_rarity)"}
"""
from __future__ import annotations
import os
from typing import Any
from scorer import load_model, score_features
class EndpointHandler:
def __init__(self, path: str = ""):
model_path = os.path.join(path, "model.json") if path else "model.json"
self.model = load_model(model_path)
self.feature_names: list[str] = self.model.get("featureNames", [])
self.expected_len = len(self.feature_names) or len(self.model["weights"])
def _score_one(self, item: dict) -> dict:
features = item.get("features")
if not isinstance(features, list) or len(features) != self.expected_len:
# Never guess: an out-of-contract input abstains completely rather than
# scoring garbage, mirroring the engine's "abstain when uncertain" rule.
return {
"score": 0.0,
"uncertainty": 1.0,
"explanation": (
f"invalid input: expected a 'features' array of length {self.expected_len} "
f"({', '.join(self.feature_names)}) -- model abstains"
),
}
out = score_features(self.model, [float(x) for x in features])
return out.to_dict()
def __call__(self, data: dict[str, Any]) -> list[dict] | dict:
inputs = data.get("inputs", data)
if isinstance(inputs, list):
return [self._score_one(item) for item in inputs]
return self._score_one(inputs)
|