FastAPI是Python生态中部署机器学习模型最平衡的选择——需用async处理I/O阻塞，复用线程池与预加载模型，结合Pydantic校验和批处理提升性能。

如何使用python中的fastapi构建高性能的机器学习模型在线服务？

直接上结论：FastAPI 是当前 Python 生态中部署机器学习模型最平衡的选择——它不强制你写异步代码，但一旦模型推理涉及 I/O（如调用外部 API、读取大文件、加载 ONNX）、或需支持高并发请求，async 就不是可选项，而是必须项。

为什么不能直接在 `def` 路由里调用 `model.predict()`

常见错误是把训练好的 scikit-learn 或 PyTorch 模型直接塞进同步函数里：

@app.post("/predict")
def predict(input: InputSchema):
    # ❌ 危险！阻塞式调用会拖垮整个事件循环
    result = model.predict([[input.x, input.y]])
    return {"result": result.item()}

问题在于：model.predict() 本身虽快，但如果模型加载路径含磁盘读取、或预处理含网络请求（如远程 embedding）、或后续要写日志到数据库，就会变成 I/O 阻塞点。FastAPI 的 ASGI 服务器（如 uvicorn）默认只开少量 worker 线程，一个阻塞请求就卡住一个线程，100 个并发可能全堵死。

PyTorch 模型若启用了 torch.set_num_threads(1)，CPU 推理反而更易被阻塞
scikit-learn 的 predict() 在多核下默认并行，但 FastAPI 同步路由无法利用异步调度优势
即使纯 CPU 推理，也要考虑请求排队时的上下文切换开销

`async` 不等于“加个 `await`”就能生效

真正起作用的是把耗时操作交给线程池或专用推理引擎，再用 await 等待结果。例如：

from concurrent.futures import ThreadPoolExecutor
import asyncio
<h1>全局线程池，避免每次新建</h1><p>executor = ThreadPoolExecutor(max_workers=4)</p><p>@app.post("/predict")
async def predict(input: InputSchema):</p><h1>✅ 正确：将 CPU 密集型任务提交到线程池</h1><pre class="brush:php;toolbar:false;">loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
    executor,
    lambda: model.predict([[input.x, input.y]])
)
return {"result": result.item()}

关键点：

ThreadPoolExecutor 必须复用，不能每次请求都 ThreadPoolExecutor() 新建
PyTorch 模型要提前调用 model.eval() 和 model.to("cpu")，否则 run_in_executor 可能触发 CUDA 上下文错误
ONNX Runtime 默认启用多线程，此时反要设 sess_options.intra_op_num_threads = 1，避免线程竞争

模型加载必须在启动时完成，而非每次请求

典型坑是把 torch.load() 或 onnxruntime.InferenceSession() 放在路由函数里：

@app.post("/infer")
async def infer(text: str):
    # ❌ 每次请求都重加载，内存爆炸 + 延迟飙升
    sess = ort.InferenceSession("model.onnx")
    ...

正确做法是用 FastAPI 的生命周期事件，在应用启动时一次性加载：

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

from fastapi import FastAPI
<p>app = FastAPI()</p><h1>全局变量存储模型</h1><p>model = None</p><p>@app.on_event("startup")
async def load_model():
global model</p><h1>加载 ONNX 模型（推荐）</h1><pre class="brush:php;toolbar:false;">model = ort.InferenceSession("model.onnx", sess_options=sess_options)
# 或加载 PyTorch 模型（注意 .eval() 和 .to()）
# model = torch.jit.load("model.pt").eval().to("cpu")

@app.post("/infer") async def infer(text: str):

✅ 直接复用已加载模型

inputs = tokenizer(text, return_tensors="pt")
...

注意：

PyTorch 的 torch.jit.script 或 torch.jit.trace 模型比 torch.load 的 state_dict 更适合服务化
ONNX 模型路径必须是绝对路径，相对路径在 uvicorn 多 worker 模式下容易出错
GPU 模型慎用：除非明确配置 uvicorn --workers 1，否则多 worker 会争抢 GPU 显存

输入验证和批处理不是可选项

生产环境里，用户传来的数据永远不可信。Pydantic 模型不只是做类型检查：

from pydantic import BaseModel, Field
<p>class PredictRequest(BaseModel):
texts: list[str] = Field(..., min_items=1, max_items=32)  # 限制 batch size
threshold: float = Field(0.5, ge=0.0, le=1.0)  # 参数范围校验</p><p>@app.post("/batch_predict")
async def batch_predict(req: PredictRequest):</p><h1>✅ 自动拒绝超长列表、越界阈值，无需手写 if 判断</h1><pre class="brush:php;toolbar:false;">results = model.predict(req.texts)
...

批处理的意义远不止吞吐量提升：