AI 模型怎么部署为本机 HTTP 服务？用 FastAPI 加载模型并验证输入

把 AI 模型部署为本机 HTTP 服务，可以用 FastAPI 在启动阶段加载模型，定义固定请求格式，再用真实 HTTP 请求检查结果和失败状态。下面只监听 127.0.0.1，演示一个 ONNX 线性打分图；它的权重由人工指定，没有训练或业务预测能力结论。

本例在 Windows、Python 3.11.15、ONNX 1.23.1、ONNX Runtime 1.28.0、NumPy 2.4.3、FastAPI 0.133.1、Pydantic 2.13.4、Uvicorn 0.41.0 环境运行。实际验证了正常请求、六类无效输入、非有限输出及缺模型启动失败；没有部署公网或测量生产并发与吞吐。

先固定服务契约

• 输入只有 features，必须包含两个有限数值；不接收模型路径、设备参数或代码。
• 模型是 score = 2×features[0] − features[1] + 0.5；score 是教学数值，不是分类概率。
• 返回 score 和 model_version，模型加载失败时直接停止启动。
• 此例用于本机联调；换成自己的模型时必须同步输入预处理、输出含义和版本记录。

FastAPI lifespan 文档展示了在请求前加载机器学习模型的模式；请求体文档说明 Pydantic 模型验证。这里进一步按Pydantic 字段规则设置严格数值、长度与有限值检查。

第一步：建立环境和模型文件

新建空练习目录并进入它，执行 Windows PowerShell 命令：

py -3.11 -m venv .venv-service
.\.venv-service\Scripts\python.exe -m pip install "onnx==1.23.1" "onnxruntime==1.28.0" "numpy==2.4.3" "fastapi==0.133.1" "pydantic==2.13.4" "uvicorn==0.41.0"

保存下方代码为 make_model.py，运行 .\.venv-service\Scripts\python.exe -X utf8 make_model.py。它在练习目录创建 service_model.onnx；图接口依据ONNX helper 文档，仅用来提供一个可复核的本地模型文件。

import numpy as np
import onnx
from onnx import TensorProto, helper, numpy_helper

graph = helper.make_graph(
    [helper.make_node('MatMul', ['features', 'weights'], ['z']),
     helper.make_node('Add', ['z', 'bias'], ['score'])],
    'linear-demo-v1',
    [helper.make_tensor_value_info('features', TensorProto.FLOAT, ['batch', 2])],
    [helper.make_tensor_value_info('score', TensorProto.FLOAT, ['batch', 1])],
    [numpy_helper.from_array(np.array([[2.0], [-1.0]], dtype=np.float32), 'weights'),
     numpy_helper.from_array(np.array([0.5], dtype=np.float32), 'bias')],
)
model = helper.make_model(graph, opset_imports=[helper.make_opsetid('', 17)])
model.ir_version = 10
onnx.checker.check_model(model)
onnx.save(model, 'service_model.onnx')
print('created: service_model.onnx')

第二步：启动时加载，请求时只做验证和推理

保存以下代码为 app.py，和 service_model.onnx 放在同一目录。ONNX Runtime Python API提供 CPU 会话和输入输出签名读取。lifespan 检查文件与签名；请求不可以自行指定其他模型路径。

from contextlib import asynccontextmanager
from pathlib import Path
from typing import Annotated
import numpy as np
import onnxruntime as ort
from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field

MODEL_VERSION = 'linear-demo-v1'
Number = Annotated[float, Field(strict=True, allow_inf_nan=False)]

class PredictInput(BaseModel):
    model_config = ConfigDict(extra='forbid')
    features: list[Number] = Field(min_length=2, max_length=2)

@asynccontextmanager
async def lifespan(app: FastAPI):
    path = Path(__file__).with_name('service_model.onnx')
    if not path.is_file():
        raise RuntimeError('服务模型文件不存在，停止启动')
    session = ort.InferenceSession(str(path), providers=['CPUExecutionProvider'])
    inputs, outputs = session.get_inputs(), session.get_outputs()
    if (len(inputs) != 1 or inputs[0].name != 'features'
        or inputs[0].type != 'tensor(float)' or inputs[0].shape != ['batch', 2]
        or len(outputs) != 1 or outputs[0].name != 'score'
        or outputs[0].type != 'tensor(float)'
        or outputs[0].shape != ['batch', 1]):
        raise RuntimeError('模型输入输出与服务契约不一致，停止启动')
    app.state.session = session
    yield
    app.state.session = None

app = FastAPI(lifespan=lifespan)

@app.exception_handler(RequestValidationError)
async def invalid_input(request: Request, error: RequestValidationError):
    # 不回显非法输入值，避免NaN等值再次破坏JSON错误响应。
    details = [{'loc': list(e['loc']), 'type': e['type'], 'message': e['msg']}
               for e in error.errors()]
    return JSONResponse(status_code=422, content={'error': 'invalid_input',
                                                  'details': details})

@app.get('/health')
def health():
    return {'status': 'ready', 'model_version': MODEL_VERSION}

@app.post('/predict')
def predict(data: PredictInput):
    with np.errstate(over='ignore', invalid='ignore'):
        x = np.array([data.features], dtype=np.float32)
    if not np.isfinite(x).all():
        raise HTTPException(status_code=422, detail='float32输入超出有限范围')
    result = app.state.session.run(['score'], {'features': x})[0]
    if result.shape != (1, 1) or not np.isfinite(result).all():
        raise HTTPException(status_code=500, detail='模型输出不符合约定')
    return {'score': float(result[0, 0]), 'model_version': MODEL_VERSION}

请求校验失败时不回显原始非法值，避免 NaN 等值再次使 JSON 错误响应序列化失败。转换成 float32 后还要检查有限值，因为原本有限的较大 Python 数值也可能在转换时溢出；模型输出同样检查，不把失败填成一个默认分数。

第三步：从终端启动本机服务

.\.venv-service\Scripts\python.exe -X utf8 -m uvicorn app:app --host 127.0.0.1 --port 8765

保持这个终端运行，另开一个 PowerShell 窗口并执行：

Invoke-RestMethod -Uri 'http://127.0.0.1:8765/health'
Invoke-RestMethod -Method Post -Uri 'http://127.0.0.1:8765/predict' -ContentType 'application/json' -Body '{"features":[3,1]}'
Invoke-RestMethod -Method Post -Uri 'http://127.0.0.1:8765/predict' -ContentType 'application/json' -Body '{"features":[3]}'

健康检查应显示 status 为 ready 和版本 linear-demo-v1。正常请求 [3,1] 的实际 HTTP 200 响应为：

{"score": 5.5, "model_version": "linear-demo-v1"}

只有一项数值的请求应返回 HTTP 422。PowerShell 可能将 422 显示为请求异常，应查看响应状态和 JSON 错误；不能把异常显示方式误判为服务没有拒绝输入。

本次还验证了哪些失败

在自己的练习中，可以先停止服务，再复制 app.py 到另一个没有模型文件的空测试目录，以相同启动命令验证失败。不要删除正在使用的模型文件来制造故障；端口已被占用和缺模型是两个不同错误。

换成真实模型时，还要带哪些资料

保留模型来源与版本、输入名/形状/dtype、预处理、输出标签或分数解释、已验证样例和依赖版本。先让实际模型在脚本里跑通固定输入，再对照同一输入的 HTTP 结果；不能只改文件名而保留这个两维输入契约。

这个回环服务没有用户鉴权、访问范围、请求限额和生产监控。若要对外提供服务，需要完成相应部署与权限设计并验证；单次本机请求成功不能作为公网服务已完成的结论。读者下一步是跑通 200、422 和启动失败，再接入自己已经验证的模型。