如何用 FastAPI 把一个 Python 算法封装成 HTTP 服务？
码海寻道 · AI 算法工程化与系统落地实战第 2 篇
算法代码要接入 Java、前端、任务队列或其他业务系统，最常见的方式是封装成 HTTP 服务。FastAPI 适合承担这一层：用 Python 类型定义请求和响应结构，用路由暴露能力，用服务层连接真正的模型推理逻辑。
本文用一个“特征分类算法”作为示例。算法本身可以替换成 OCR、图像分类、目标检测、文本分类或预测模型，接口设计方法基本一致。
一、先确定服务边界
不要一上来就在路由函数里写完整算法。先明确四件事：
HTTP 层：接收请求、校验参数、返回状态码 业务层：编排输入处理、推理和结果保存 推理层：加载模型、执行算法、返回结构化结果 基础设施层：数据库、对象存储、日志和队列调用链可以设计成：
客户端 ↓ HTTP/JSON 或 multipart/form-data FastAPI Router ↓ 参数校验与鉴权 Algorithm Service ↓ 获取模型实例 Inference Engine ↓ 结构化响应这样做的好处是，未来把同步推理改成 RabbitMQ Worker 时，推理层不需要重写。
二、创建最小项目
algorithm-api/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── schemas.py │ ├── inference.py │ └── routes.py ├── tests/ ├── requirements.txt └── Dockerfile依赖可以先保持简单：
fastapi uvicorn[standard] pydantic-settings如果接收文件，还要增加python-multipart。如果算法依赖 NumPy、ONNX Runtime 或 PyTorch，再按实际运行时加入，不要一开始把所有机器学习库都放进依赖文件。
三、定义请求和响应模型
# app/schemas.pyfrompydanticimportBaseModel,FieldclassPredictRequest(BaseModel):features:list[float]=Field(min_length=1,max_length=1024)model_version:str|None=NoneclassPrediction(BaseModel):label:strscore:float=Field(ge=0,le=1)classPredictResponse(BaseModel):request_id:strstatus:strmodel_version:strresult:Prediction elapsed_ms:int请求模型负责校验输入，响应模型负责约束输出。不要让算法函数直接返回任意字典，否则模型升级后多出或少了字段，调用方可能只能在运行时才发现问题。
四、把模型加载放到应用生命周期
# app/inference.pyclassClassifier:def__init__(self,model_path:str,version:str):self.model_path=model_path self.version=version self.model=self._load_model(model_path)def_load_model(self,model_path:str):# 真实项目中替换成 ONNX Runtime、PyTorch 或其他推理引擎return{"path":model_path}defpredict(self,features:list[float])->dict:score=min(sum(features)/len(features),1.0)return{"label":"positive"ifscore>=0.5else"negative","score":score}在应用启动时构造模型实例：
# app/main.pyfromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIfromapp.configimportsettingsfromapp.inferenceimportClassifier@asynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.classifier=Classifier(model_path=settings.model_path,version=settings.model_version,)yieldapp.state.classifier=Noneapp=FastAPI(title="Algorithm API",version="1.0.0",lifespan=lifespan)如果模型加载失败，应用应该启动失败或保持未就绪，而不是让第一条用户请求承担初始化错误。对于 GPU 模型，还要在启动阶段执行一次预热，提前发现 CUDA、显存或模型文件问题。
五、实现预测接口
# app/routes.pyfromtimeimportperf_counterfromuuidimportuuid4fromfastapiimportAPIRouter,HTTPException,Requestfromapp.schemasimportPredictRequest,PredictResponse,Prediction router=APIRouter(prefix="/api/v1",tags=["algorithm"])@router.post("/predict",response_model=PredictResponse)defpredict(payload:PredictRequest,request:Request)->PredictResponse:request_id=request.headers.get("X-Request-ID",str(uuid4()))started=perf_counter()classifier=getattr(request.app.state,"classifier",None)ifclassifierisNone:raiseHTTPException(status_code=503,detail="MODEL_NOT_READY")try:result=classifier.predict(payload.features)exceptValueErrorasexc:raiseHTTPException(status_code=422,detail="INVALID_INPUT")fromexcexceptExceptionasexc:# 真实项目应记录完整异常，但不要把堆栈直接返回客户端request.app.logger.error("inference_failed: %s",exc)raiseHTTPException(status_code=500,detail="INFERENCE_FAILED")fromexcreturnPredictResponse(request_id=request_id,status="succeeded",model_version=classifier.version,result=Prediction(**result),elapsed_ms=round((perf_counter()-started)*1000),)最后在main.py注册路由：
fromapp.routesimportrouter app.include_router(router)六、文件型算法如何接入？
OCR、图像分类和目标检测通常接收文件：
fromfastapiimportAPIRouter,File,UploadFile router=APIRouter()@router.post("/predict-file")asyncdefpredict_file(file:UploadFile):iffile.content_typenotin{"image/jpeg","image/png"}:return{"status":"rejected","code":"UNSUPPORTED_FILE_TYPE"}content=awaitfile.read()iflen(content)>10*1024*1024:return{"status":"rejected","code":"FILE_TOO_LARGE"}# 真实项目中还需要校验文件内容、病毒扫描和临时文件生命周期result=run_image_inference(content)return{"status":"succeeded","filename":file.filename,"result":result}不要只相信扩展名或客户端传来的 MIME 类型。生产环境要检查文件大小、真实格式、图片尺寸、临时文件路径和对象存储权限；大文件则建议先上传对象存储，再让算法服务读取受控的对象地址。
七、同步接口还是异步任务？
可以用一个简单标准判断：
几十毫秒到几秒内完成：同步 HTTP 处理时间不稳定或超过请求超时：异步任务 需要批量处理大量文件：消息队列 + Worker同步接口适合在线推荐、单张图片判断和短文本分类。OCR 大文件、视频分析、批量预测则更适合返回job_id，由客户端查询进度。不要为了“接口看起来简单”把长任务一直占着 HTTP 连接。
八、如何运行和验证？
uvicorn app.main:app--reload--port8000请求测试：
curl-XPOST"http://localhost:8000/api/v1/predict"`-H"Content-Type: application/json"`-H"X-Request-ID: req-demo-001"`-d'{"features":[0.2,0.8,0.6]}'验证时至少检查：
- 正常输入是否返回稳定 Schema；
- 空数组、超长数组和错误类型是否被拒绝；
- 模型未加载时是否返回 503；
- 推理异常是否不会泄露堆栈；
- 请求 ID 是否出现在响应和日志中；
- 多次请求是否复用同一个模型实例。
结语
FastAPI 的价值不只是快速写出一个路由，而是帮助算法团队建立清晰的服务契约：输入可校验、输出可预测、模型可复用、错误可处理、版本可追踪。完成这一层之后，Docker 才能把它稳定地交付到其他环境。
下一篇将把这个 FastAPI 服务构建成 Docker 镜像。
参考资料
- FastAPI 官方文档：Request Body
- FastAPI 官方文档：Response Model
- FastAPI 官方文档：Request Files
本文为“码海寻道”原创技术文章。接口示例中的推理函数是演示实现，生产项目应替换为实际模型，并补充鉴权、限流、审计和数据安全处理。
