使用 FastAPI 将模型封装为在线的 API 服务

从离线模型到在线API服务的解决方案之一

目标

将本地/离线模型封装为可调用的 HTTP API 服务，统一输入输出契约，支持训练、预测、状态与模型管理，便于前端/平台集成。

设计要点

快速模板

1. 项目结构

app/
  main.py              # 启动入口（聚合 routers）
  api/
    __init__.py
    routers/
      predict.py       # 预测路由
      train.py         # 训练路由
      health.py        # 健康检查
    service/
      predictor.py     # 预测编排
      trainer.py       # 训练编排
  core/
    model.py           # 模型定义/加载/保存
    io.py              # 标准化/反标准化/序列构造

2. 依赖声明（示例）

pip install fastapi uvicorn pydantic numpy torch scikit-learn joblib

3. 关键代码片段

main.py

from fastapi import FastAPI
from api.routers.predict import router as predict_router
from api.routers.train import router as train_router
from api.routers.health import router as health_router

app = FastAPI(title="模型服务", version="1.0.0")
app.include_router(train_router, prefix="/train")
app.include_router(predict_router, prefix="/predict")
app.include_router(health_router, prefix="")

预测路由（最小示例）

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List
from api.service.predictor import predict_batch

router = APIRouter(tags=["预测"])

class PredictIn(BaseModel):
  future_features: List[List[float]]
  recent_historical_features: List[List[float]]
  recent_historical_target: List[float]
  model_id: str | None = None

@router.post("", summary="批量预测")
async def do_predict(payload: PredictIn):
  try:
    return await predict_batch(payload)
  except Exception as e:
    raise HTTPException(status_code=500, detail=str(e))

训练路由（异步）

from fastapi import APIRouter, BackgroundTasks
from pydantic import BaseModel
from api.service.trainer import start_training

router = APIRouter(tags=["训练"])

class TrainIn(BaseModel):
  historical_target: list[float]
  historical_features: list[list[float]]
  sequence_length: int = 30

@router.post("", summary="启动训练（异步）")
async def do_train(payload: TrainIn, background: BackgroundTasks):
  task_id = await start_training(payload, background)
  return {"task_id": task_id, "status": "accepted"}

健康检查

from fastapi import APIRouter
router = APIRouter(tags=["健康检查"])

@router.get("/health")
async def health():
  return {"status": "healthy"}

关键

常见问题（FAQ）