FastAPI处理表单数据预测时出现442 Unprocessable Entity错误求助

错误原因分析 #

1. Pydantic模型未转为字典：代码中data_dict = data直接赋值Pydantic实例，pd.DataFrame.from_dict无法识别Pydantic对象结构，导致DataFrame构造失败
2. numpy数组无法序列化：LightGBM预测返回的是numpy数组，FastAPI默认JSON序列化器无法处理该类型
3. 文件操作不规范：未使用上下文管理器with处理pickle文件，可能导致资源泄漏
4. 表单数据适配缺失：当前接口仅支持JSON请求，若要接收表单数据需额外配置

修正后的完整代码 #

import pandas as pd
import pickle
from fastapi import FastAPI, Form
from pydantic import BaseModel
from typing import Annotated

app = FastAPI(title='Placement Analytics', version='1.0',
              description='Lightgbm model is used for prediction')

class Data(BaseModel):
    oldRoad: float 
    onRoadNow: float 
    years: float
    km: float
    rating: float
    condition: float
    economy: float
    topSpeed: float
    horsePower: float
    torque: float

# 支持JSON请求的预测接口
@app.post("/predict/json")
def predict_json(data: Data):
    # 将Pydantic模型转为字典（Pydantic v2用model_dump，v1用dict()）
    data_dict = data.model_dump()
    df2 = pd.DataFrame.from_dict([data_dict])
    
    # 用上下文管理器安全加载模型
    with open("predict.pkl","rb") as pickle_in:
        classifier = pickle.load(pickle_in)
    
    prediction = classifier.predict(df2)
    # 将numpy数组转为Python列表，支持JSON序列化
    return {"prediction": prediction.tolist()}

# 支持表单数据的预测接口
@app.post("/predict/form")
def predict_form(
    oldRoad: Annotated[float, Form()],
    onRoadNow: Annotated[float, Form()],
    years: Annotated[float, Form()],
    km: Annotated[float, Form()],
    rating: Annotated[float, Form()],
    condition: Annotated[float, Form()],
    economy: Annotated[float, Form()],
    topSpeed: Annotated[float, Form()],
    horsePower: Annotated[float, Form()],
    torque: Annotated[float, Form()]
):
    # 构造字典并转为DataFrame
    data_dict = {
        "oldRoad": oldRoad,
        "onRoadNow": onRoadNow,
        "years": years,
        "km": km,
        "rating": rating,
        "condition": condition,
        "economy": economy,
        "topSpeed": topSpeed,
        "horsePower": horsePower,
        "torque": torque
    }
    df2 = pd.DataFrame.from_dict([data_dict])
    
    with open("predict.pkl","rb") as pickle_in:
        classifier = pickle.load(pickle_in)
    
    prediction = classifier.predict(df2)
    return {"prediction": prediction.tolist()}

@app.get('/home')
def read_home():
    """
    Home endpoint which can be used to test the availability of the application.
    """
    return {'message': 'System is healthy'}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run("main:app", host="127.0.0.1",port=8000, reload=True, debug=True)

关键修改说明 #

• 将Pydantic实例转为字典，确保DataFrame能正确构造
• 用with语句处理pickle文件，自动释放文件资源
• 将预测结果转为Python列表，解决JSON序列化问题
• 新增表单数据专属接口，同时保留JSON请求支持，覆盖不同提交场景

内容的提问来源于stack exchange，提问作者ABUBAKAR MUHAMMED MUKTAR