Biến Mô Hình Phân Tích Thành API Chuyên Nghiệp Với Ngôn Ngữ Python Và FastAPI

pip install fastapi uvicorn pydantic pandas scikit-learn

my_analysis_api/
│
├── main.py
└── requirements.txt

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import pandas as pd

# Khởi tạo ứng dụng FastAPI với tiêu đề và mô tả SEO
app = FastAPI(
    title="Analysis Model API",
    description="API đóng gói mô hình phân tích dữ liệu sử dụng Python và FastAPI",
    version="1.0.0"
)

# Định nghĩa hằng số cho phiên bản cấu hình mô hình
MODEL_VERSION = "v1.0.2"

# Định nghĩa cấu trúc dữ liệu đầu vào sử dụng Pydantic
class AnalysisInput(BaseModel):
    feature_a: float
    feature_b: float
    category: str

# Định nghĩa cấu trúc dữ liệu đầu ra chuẩn hóa
class AnalysisOutput(BaseModel):
    prediction_score: float
    status: str
    model_version: str

def run_analysis_model(data: AnalysisInput) -> float:
    """
    Hàm thực thi logic phân tích dữ liệu dựa trên thông số đầu vào của người dùng.
    Trong thực tế, bạn sẽ tải mô hình đã huấn luyện từ file .pkl, .h5 hoặc .onnx tại đây.
    """
    # Giả lập công thức tính toán trọng số phân tích
    base_score = (data.feature_a * 1.5) + (data.feature_b * 2.0)
    
    # Điều chỉnh điểm số dự đoán dựa trên phân loại danh mục (category)
    if data.category.lower() == "premium":
        base_score *= 1.2
    else:
        base_score *= 0.9
        
    return round(base_score, 2)

@app.post("/predict", response_model=AnalysisOutput)
def predict_endpoint(payload: AnalysisInput):
    """
    Điểm cuối (endpoint) nhận phương thức POST, xử lý dữ liệu và trả về kết quả phân tích.
    """
    try:
        # Gọi hàm xử lý cốt lõi của mô hình phân tích
        calculated_score = run_analysis_model(payload)
        
        # Đóng gói kết quả trả về theo đúng lược đồ Pydantic đã định nghĩa
        response_data = AnalysisOutput(
            prediction_score=calculated_score,
            status="success",
            model_version=MODEL_VERSION
        )
        return response_data
        
    except Exception as error:
        # Bắt ngoại lệ và trả về mã lỗi HTTP 500 nếu quá trình tính toán gặp sự cố
        raise HTTPException(status_code=500, detail=str(error))

uvicorn main:app --reload

from typing import List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="Batch Analysis API")

class AnalysisInput(BaseModel):
    feature_a: float
    feature_b: float
    category: str

# Định nghĩa cấu trúc đầu vào chứa danh sách các bản ghi
class BatchInput(BaseModel):
    items: List[AnalysisInput]

class BatchOutput(BaseModel):
    results: List[float]
    total_processed: int
    status: str

def run_analysis_model(data: AnalysisInput) -> float:
    """Hàm tính toán điểm số cho từng bản ghi đơn lẻ."""
    base_score = (data.feature_a * 1.5) + (data.feature_b * 2.0)
    if data.category.lower() == "premium":
        base_score *= 1.2
    return round(base_score, 2)

@app.post("/predict-batch", response_model=BatchOutput)
def predict_batch_endpoint(payload: BatchInput):
    """
    Điểm cuối xử lý mảng dữ liệu đầu vào và trả về kết quả dự đoán hàng loạt.
    """
    try:
        # Sử dụng danh sách hiểu (list comprehension) để duyệt qua từng phần tử
        calculated_results = [run_analysis_model(item) for item in payload.items]
        
        return BatchOutput(
            results=calculated_results,
            total_processed=len(calculated_results),
            status="success"
        )
    except Exception as error:
        raise HTTPException(status_code=500, detail=str(error))

import io
from fastapi import FastAPI, File, UploadFile, HTTPException
import pandas as pd

app = FastAPI(title="CSV Analysis File API")

@app.post("/predict-csv")
async def predict_csv_endpoint(file: UploadFile = File(...)):
    """
    Nhận tệp CSV từ client, đọc bằng Pandas và trả về điểm số trung bình phân tích.
    """
    # Kiểm tra định dạng tệp gửi lên có phải là CSV không
    if not file.filename.endswith('.csv'):
        raise HTTPException(status_code=400, detail="Chỉ hỗ trợ tải lên tệp định dạng .csv")
    
    try:
        # Đọc nội dung tệp bất đồng bộ
        content = await file.read()
        
        # Chuyển đổi dữ liệu nhị phân thành DataFrame của Pandas
        dataframe = pd.read_csv(io.BytesIO(content))
        
        # Kiểm tra sự tồn tại của các cột bắt buộc trong tệp
        required_columns = ["feature_a", "feature_b"]
        for column in required_columns:
            if column not in dataframe.columns:
                raise HTTPException(status_code=400, detail=f"Tệp thiếu cột bắt buộc: {column}")
                
        # Thực hiện tính toán phân tích hàng loạt trên toàn bộ DataFrame
        dataframe["predicted_score"] = (dataframe["feature_a"] * 1.5) + (dataframe["feature_b"] * 2.0)
        
        # Tính toán giá trị trung bình kết quả để trả về thống kê nhanh
        mean_score = float(dataframe["predicted_score"].mean())
        total_rows = len(dataframe)
        
        return {
            "filename": file.filename,
            "total_rows_processed": total_rows,
            "average_predicted_score": round(mean_score, 2),
            "status": "success"
        }
        
    except Exception as error:
        raise HTTPException(status_code=500, detail=f"Lỗi xử lý tệp: {str(error)}")