การให้บริการการทำนายด้วยปลายทาง FastAPI

# pip install fastapi uvicorn[standard]

# Verify installation
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

print('FastAPI ready')

from pydantic import BaseModel
from typing import Optional

class IrisFeatures(BaseModel):
    sepal_length: float
    sepal_width: float
    petal_length: float
    petal_width: float

class PredictionResponse(BaseModel):
    predicted_class: int
    class_name: str
    confidence: float

# Example input (FastAPI will validate this automatically)
input_data = IrisFeatures(sepal_length=5.1, sepal_width=3.5,
                           petal_length=1.4, petal_width=0.2)
print('Input:', input_data)

import joblib
from contextlib import asynccontextmanager
from fastapi import FastAPI

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: load model once
    ml_models['iris'] = joblib.load('/tmp/iris_pipeline.joblib')
    print('Model loaded at startup')
    yield
    # Shutdown: cleanup if needed
    ml_models.clear()

app = FastAPI(title='Iris Predictor API', lifespan=lifespan)

import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel
import joblib

app = FastAPI()
model = None

@app.on_event('startup')
def load_model():
    global model
    model = joblib.load('/tmp/iris_pipeline.joblib')

CLASS_NAMES = ['setosa', 'versicolor', 'virginica']

@app.post('/predict')
def predict(features: IrisFeatures):
    X = np.array([[features.sepal_length, features.sepal_width,
                   features.petal_length, features.petal_width]])
    pred = int(model.predict(X)[0])
    proba = float(model.predict_proba(X)[0].max())
    return {
        'predicted_class': pred,
        'class_name': CLASS_NAMES[pred],
        'confidence': round(proba, 4)
    }

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get('/health')
def health():
    if model is None:
        return JSONResponse(status_code=503,
                            content={'status': 'unhealthy', 'reason': 'model not loaded'})
    return {'status': 'ok', 'model': 'iris_pipeline', 'version': '1.0.0'}

@app.get('/')
def root():
    return {'message': 'Iris Predictor API — POST /predict to get a classification'}

# Save to main.py then run:
# uvicorn main:app --host 0.0.0.0 --port 8000 --reload

# Test with curl:
# curl -X POST http://localhost:8000/predict \
#   -H 'Content-Type: application/json' \
#   -d '{"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}'
#
# Expected response:
# {"predicted_class": 0, "class_name": "setosa", "confidence": 0.9981}

print('Command to start: uvicorn main:app --reload --port 8000')

import requests

url = 'http://localhost:8000/predict'
payload = {
    'sepal_length': 6.3,
    'sepal_width': 3.3,
    'petal_length': 6.0,
    'petal_width': 2.5
}

response = requests.post(url, json=payload)
if response.status_code == 200:
    result = response.json()
    print('Predicted class:', result['class_name'])
    print('Confidence:', result['confidence'])
else:
    print('Error:', response.status_code, response.text)

from fastapi import FastAPI, HTTPException
import numpy as np

app = FastAPI()

@app.post('/predict')
def predict(features: IrisFeatures):
    try:
        X = np.array([[features.sepal_length, features.sepal_width,
                       features.petal_length, features.petal_width]])
        if np.any(np.isnan(X)) or np.any(X < 0):
            raise HTTPException(status_code=400,
                                detail='Input contains invalid values (NaN or negative)')
        pred = int(model.predict(X)[0])
        proba = float(model.predict_proba(X)[0].max())
        return {'predicted_class': pred, 'confidence': round(proba, 4)}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail='Internal prediction error')

from typing import List
from pydantic import BaseModel
import numpy as np

class BatchRequest(BaseModel):
    instances: List[IrisFeatures]

@app.post('/predict/batch')
def predict_batch(batch: BatchRequest):
    X = np.array([[f.sepal_length, f.sepal_width, f.petal_length, f.petal_width]
                  for f in batch.instances])
    preds = model.predict(X).tolist()
    probas = model.predict_proba(X).max(axis=1).tolist()
    return {'predictions': [{'class': p, 'confidence': round(c, 4)}
                            for p, c in zip(preds, probas)]}