산업 현장 AI 통합 아키텍처 가이드 3편: 엣지-클라우드 하이브리드 배포

import onnxruntime as ort
import numpy as np
import time

class FastInference:
    def __init__(self, model_path: str):
        opts = ort.SessionOptions()
        opts.intra_op_num_threads = 4
        opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        self.sess = ort.InferenceSession(model_path, sess_options=opts)
        self.inp = self.sess.get_inputs()[0].name
        # 웜업: 첫 추론은 JIT 컴파일로 느림
        dummy = np.zeros((1, 128), dtype=np.float32)
        for _ in range(10):
            self.sess.run(None, {self.inp: dummy})

    def predict(self, features: np.ndarray) -> dict:
        t0 = time.perf_counter()
        out = self.sess.run(None, {self.inp: features})
        return {
            "value": float(out[0][0]),
            "latency_ms": (time.perf_counter() - t0) * 1000
        }

import asyncio
from collections import deque

class BatchInference:
    def __init__(self, model, batch_size=32, max_wait_ms=10):
        self.model = model
        self.batch_size = batch_size
        self.max_wait = max_wait_ms / 1000
        self.queue = deque()
        asyncio.create_task(self._loop())

    async def predict(self, features):
        fut = asyncio.Future()
        self.queue.append((features, fut))
        return await fut

    async def _loop(self):
        while True:
            await asyncio.sleep(self.max_wait)
            if not self.queue:
                continue
            batch, futs = [], []
            while self.queue and len(batch) < self.batch_size:
                f, fut = self.queue.popleft()
                batch.append(f)
                futs.append(fut)
            results = self.model.predict_batch(np.stack(batch))
            for fut, res in zip(futs, results):
                fut.set_result(res)

from sklearn.ensemble import IsolationForest
import joblib

# 정상 운전 데이터만으로 학습
clf = IsolationForest(n_estimators=100, contamination=0.01, random_state=42)
clf.fit(normal_sensor_data)
joblib.dump(clf, "pdm_model.pkl")

def score(reading: np.ndarray) -> dict:
    s = clf.decision_function(reading.reshape(1, -1))[0]
    return {
        "score": float(s),
        "is_anomaly": s < -0.1,
        "severity": "HIGH" if s < -0.3 else "MEDIUM" if s < -0.1 else "NORMAL"
    }

from scipy.stats import ks_2samp

class DriftDetector:
    def __init__(self, reference: np.ndarray, threshold: float = 0.05):
        self.ref = reference
        self.thr = threshold

    def check(self, current: np.ndarray) -> list:
        drifted = []
        for i in range(self.ref.shape[1]):
            _, p = ks_2samp(self.ref[:, i], current[:, i])
            if p < self.thr:
                drifted.append(f"feature_{i}")
        if drifted:
            print(f"드리프트 감지: {drifted}")
            self._trigger_retrain()
        return drifted

    def _trigger_retrain(self):
        import requests
        requests.post(os.environ["RETRAIN_WEBHOOK_URL"])

# 1) 배포된 모델 무결성: 클라우드에서 발행한 해시와 비교
sha256sum /opt/models/current/model.onnx
# 2) 런타임과 실행 프로바이더 확인
python3 -c "import onnxruntime as ort; print(ort.__version__, ort.get_available_providers())"
# 3) 모델 opset 확인
python3 -c "import onnx; m=onnx.load('/opt/models/current/model.onnx'); print([(o.domain,o.version) for o in m.opset_import])"