Spaces:
Running
Running
File size: 3,807 Bytes
53d0a86 | 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 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """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()
|