Spaces:
Running
Running
| """One persistent native model, explicit offline state, no synthetic predictions.""" | |
| import os | |
| import sys | |
| import threading | |
| import time | |
| from pathlib import Path | |
| from contract import MODEL, to_records, to_response, timing_shape | |
| DEFAULT_MANIFEST = "da603662bc57e89ccfb51c972ed9c1f2825f267597353cf1337df9117a3dfabe" | |
| class Unavailable(RuntimeError): | |
| pass | |
| class Busy(RuntimeError): | |
| pass | |
| class Engine: | |
| def __init__(self): | |
| self.model = os.getenv("DECISION_MODEL_ID", MODEL) | |
| self.native = None | |
| self.lock = threading.Lock() | |
| self.phase = "not_loaded" | |
| self.native_path = os.getenv("DECISION_NATIVE", "") | |
| self.manifest = os.getenv("DECISION_MANIFEST_SHA256", DEFAULT_MANIFEST) | |
| def status(self): | |
| return {"configured": bool(self.native_path), "loaded": self.native is not None, | |
| "phase": self.phase if self.native_path else "offline", "model": self.model, | |
| "manifest_sha256": self.manifest, "complete_input_tokens": 1024, | |
| "context_batch": True, "backend": "native_rocm", "live_inference": self.native is not None} | |
| def _load(self): | |
| if self.native is not None: | |
| return 0.0 | |
| if not self.native_path: | |
| raise Unavailable("No model is connected. Set DECISION_NATIVE to a verified native export on the AMD server.") | |
| runtime = Path(os.getenv("DECISION_RUNTIME_PATH", str(Path(__file__).resolve().parents[1] / "release"))) | |
| if str(runtime) not in sys.path: | |
| sys.path.insert(0, str(runtime)) | |
| started = time.perf_counter() | |
| self.phase = "loading" | |
| import torch | |
| from decision_runtime import load_native | |
| if torch.version.hip is None or not torch.cuda.is_available() or torch.cuda.device_count() != 1: | |
| raise Unavailable("Expose exactly one AMD ROCm GPU for the native preview. CPU fallback is not enabled.") | |
| torch.cuda.set_device(0) | |
| torch.set_num_threads(2) | |
| torch.backends.cuda.matmul.allow_tf32 = False | |
| torch.backends.cudnn.allow_tf32 = False | |
| torch.backends.mha.set_fastpath_enabled(False) | |
| self.native = load_native(self.native_path, expected_manifest_sha256=self.manifest, device="cuda:0") | |
| self.phase = "ready" | |
| return (time.perf_counter() - started) * 1000 | |
| def evaluate(self, body, records): | |
| if body.get("model", MODEL) != self.model: | |
| raise ValueError("Request model does not match this worker") | |
| if records != to_records(body, model=self.model): | |
| raise ValueError("Request records changed") | |
| if not self.lock.acquire(blocking=False): | |
| raise Busy("The model is finishing another request. Try again in a moment.") | |
| started = time.perf_counter() | |
| try: | |
| load_ms = self._load() | |
| import torch | |
| from decision_inference._grouped import predict_grouped_1k as predict_1k | |
| self.phase = "running" | |
| torch.cuda.synchronize() | |
| begin = time.perf_counter() | |
| predictions = predict_1k(self.native, records, batch_size=8) | |
| torch.cuda.synchronize() | |
| inference_ms = (time.perf_counter() - begin) * 1000 | |
| result = to_response(body, records, predictions, model=self.model) | |
| result["timing"] = {"model_load_ms": round(load_ms, 2), "inference_ms": round(inference_ms, 2), | |
| "server_ms": round((time.perf_counter() - started) * 1000, 2), | |
| "cold_start": load_ms > 0, **timing_shape(body, records)} | |
| result["source"] = "live_native" | |
| return result | |
| finally: | |
| self.phase = "ready" if self.native is not None else "not_loaded" | |
| self.lock.release() | |