FastAPI机器学习部署脚本返回Null结果排查求助

问题描述 #

开发基于FastAPI的二元分类机器学习服务，本地训练的流水线功能正常，但部署后接口始终返回Null（预期输出0或1）。代码如下：

from pydantic import BaseModel
from fastapi import FastAPI
import pandas as pd
import joblib

import uvicorn


# instance
app = FastAPI()


# data model
class DataModel(BaseModel):
    id: str
    radius_mean: float
    ...
    # all other features, except for target variable that has to be predicted

class ItemOut(BaseModel):
    id: str
    diagnosis: int


def load_weights():
    model = joblib.load("models/RF/RF_weights_v1.0.pkl")
    return model


def predict(data):
    model = load_weights()
    preprocessing = model.named_steps["preprocessing"]
    transformed_data = preprocessing.transform(data)
    classifier = model.named_steps["classifier"]
    predicted_values = classifier.predict(transformed_data)
    return predicted_values


@app.post("/predict", response_model=ItemOut)
async def predict(dataModel: DataModel):
    data = pd.DataFrame([dataModel.dict()])
    predictions = predict(data)
    return {"id": "id", "diagnosis": predictions}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.1", port=8000)

问题原因 #

1. 返回值类型不匹配：模型predict方法返回的是一维NumPy数组（如array([0])），但ItemOut中diagnosis定义为int类型，Pydantic无法将数组解析为整数，导致验证失败，最终返回Null。
2. ID硬编码错误：返回的id被固定为字符串"id"，未使用传入的dataModel.id，但这不是返回Null的直接原因。
3. 模型重复加载与错误无捕获：每次请求都重新加载模型，不仅效率低下，若模型路径错误或加载失败，会直接抛出异常且无处理，导致接口返回错误而非预期结果。
4. Host配置错误：0.0.0.1仅允许本地回环访问，外部请求无法正常连接，但这会导致请求失败而非返回Null。

修复代码 #

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

# 全局加载模型，避免重复IO操作
app = FastAPI()
MODEL = None

@app.on_event("startup")
async def load_model_on_startup():
    global MODEL
    try:
        MODEL = joblib.load("models/RF/RF_weights_v1.0.pkl")
    except Exception as e:
        raise RuntimeError(f"Failed to load model: {str(e)}")

class DataModel(BaseModel):
    id: str
    radius_mean: float
    # 补充其他特征定义...

class ItemOut(BaseModel):
    id: str
    diagnosis: int

def predict(data):
    if MODEL is None:
        raise RuntimeError("Model not loaded")
    try:
        transformed_data = MODEL.named_steps["preprocessing"].transform(data)
        predicted_values = MODEL.named_steps["classifier"].predict(transformed_data)
        # 取出单样本预测值并转为int类型
        return int(predicted_values[0])
    except Exception as e:
        raise ValueError(f"Prediction failed: {str(e)}")

@app.post("/predict", response_model=ItemOut)
async def predict_endpoint(dataModel: DataModel):
    try:
        data = pd.DataFrame([dataModel.dict()])
        diagnosis = predict(data)
        return {"id": dataModel.id, "diagnosis": diagnosis}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

部署最佳实践 #

• 模型预加载：在服务启动时通过startup事件加载模型到全局变量，避免每次请求重复加载，提升性能并减少IO错误。
• 支持批量预测：允许传入多个样本，提升服务吞吐量，示例代码如下：

  from typing import List
  
  @app.post("/predict/batch", response_model=List[ItemOut])
  async def predict_batch(dataModels: List[DataModel]):
      try:
          data = pd.DataFrame([item.dict() for item in dataModels])
          transformed_data = MODEL.named_steps["preprocessing"].transform(data)
          predicted_values = MODEL.named_steps["classifier"].predict(data)
          return [{"id": item.id, "diagnosis": int(pred)} for item, pred in zip(dataModels, predicted_values)]
      except Exception as e:
          raise HTTPException(status_code=400, detail=str(e))
  

• 严格数据验证：利用Pydantic模型确保输入特征的类型、范围符合要求，避免脏数据导致预测失败。
• 错误处理与日志：添加try-except捕获异常，返回明确的HTTP错误码和信息；配置日志记录请求参数、预测结果和错误详情，方便排查问题。
• 环境配置：用环境变量管理模型路径、服务端口、日志级别等参数，避免硬编码，提升部署灵活性。
• 生产级部署：使用gunicorn作为进程管理器搭配uvicorn工作进程，命令示例：

  gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
  

• 模型版本管理：为模型文件添加版本标识（如RF_weights_v1.0.pkl），部署时明确指定版本，方便回滚和跟踪模型迭代。

内容的提问来源于stack exchange，提问作者Simone De Bellis