【量子AI #18】FastAPIでQMLモデルをREST APIとしてサービングする方法

7. 本番システムへの統合パターン

7.1 QMLサービスのREST API化

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import torch, numpy as np, asyncio, time, logging
from datetime import datetime

app = FastAPI(title="製造品質予測 QML API", version="2.0.0")

class SensorReading(BaseModel):
    sensor_values: list = Field(..., min_items=32, max_items=32)
    machine_id: str
    batch_id: str = None
    timestamp: datetime = None

class QualityPrediction(BaseModel):
    status: str           # "normal" or "defective"
    defect_probability: float
    normal_probability: float
    confidence: str       # "high", "medium", "low"
    quantum_expectation_values: list
    processing_time_ms: float
    recommendation: str

class QMLInferenceEngine:
    def __init__(self):
        self.model = HybridQNN(n_features=32)
        self.scaler = StandardScaler()
        self.scaler.mean_ = np.zeros(32)
        self.scaler.scale_ = np.ones(32)
        self.model.eval()
        self._lock = asyncio.Lock()
        self._count = 0

    async def predict(self, sensor_values):
        async with self._lock:  # 量子回路は排他制御
            t0 = time.perf_counter()
            X = np.array(sensor_values).reshape(1,-1)
            X_s = self.scaler.transform(X)
            with torch.no_grad():
                probs = self.model(torch.FloatTensor(X_s))
                qinfo = self.model.get_quantum_state_info(torch.FloatTensor(X_s))
            dp = float(probs[0,1]); np_ = float(probs[0,0])
            status = "defective" if dp>0.5 else "normal"
            conf = "high" if max(dp,np_)>0.85 else ("medium" if max(dp,np_)>0.7 else "low")
            rec = "即時ライン停止" if dp>0.8 else ("詳細検査推奨" if status=="defective" else "合格・出荷可")
            ms = (time.perf_counter()-t0)*1000; self._count+=1
            return {"status":status,"defect_probability":dp,"normal_probability":np_,
                    "confidence":conf,"quantum_expectation_values":qinfo["quantum_expectation_values"][0].tolist(),
                    "processing_time_ms":ms,"recommendation":rec}