008、模型部署实战Python用OpenCV和FastAPI搭个推理服务昨天深夜调试一个边缘设备上的模型服务发现请求延迟忽高忽低。用curl压测时偶尔会出现响应时间从50ms突然跳到500ms的情况。打开htop一看CPU占用率并不高内存也充足。问题出在哪后来发现是OpenCV的DNN模块在加载模型时默认配置对并发请求支持不够友好。这个坑让我决定整理一下实际可用的部署方案。一、环境准备与模型转换先说说模型格式。YOLOv11训练出来的通常是.pt文件部署时得转成OpenCV能读的格式。这里推荐转成ONNXOpenCV的DNN模块对ONNX支持比较稳定。# export_onnx.pyimporttorchfrommodels.yoloimportModel# 加载训练好的权重modelModel(yolov11s.yaml)# 根据你的配置文件来model.load_state_dict(torch.load(yolov11s.pt)[model].float().state_dict())model.eval()# 准备一个示例输入dummy_inputtorch.randn(1,3,640,640)# 导出ONNX - 这里有个细节要注意torch.onnx.export(model,dummy_input,yolov11s.onnx,opset_version12,# 别用太新的版本OpenCV可能不支持input_names[images],output_names[output],dynamic_axes{images:{0:batch_size}}# 支持动态batch)转换完记得验证一下输出是否一致。我遇到过ONNX转换后输出维度对不上的情况特别是当模型里有自定义层的时候。二、OpenCV推理引擎封装直接上代码重点看注释里的坑点。# inference_engine.pyimportcv2importnumpyasnpfromtypingimportList,TupleimporttimeclassYOLOv11Engine:def__init__(self,onnx_path:str,conf_thresh:float0.5): 初始化推理引擎 onnx_path: ONNX模型路径 conf_thresh: 置信度阈值默认0.5够用了 self.netcv2.dnn.readNetFromONNX(onnx_path)# 重要设置后端和目标设备# 如果用CPU建议用OPENBLAS比默认的EIGEN快self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_OPENCV)self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU)# 如果是英伟达显卡且编译了CUDA支持# self.net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)# self.net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)self.conf_thresholdconf_thresh self.input_size(640,640)# 根据你的模型输入尺寸调整# 预热一下避免第一次推理慢dummynp.zeros((1,3,*self.input_size),dtypenp.float32)self.net.setInput(dummy)_self.net.forward()print(f引擎初始化完成输入尺寸:{self.input_size})defpreprocess(self,image:np.ndarray)-np.ndarray: 预处理resize 归一化 转BGR 注意OpenCV默认读进来是BGR训练时通常是RGB # 保持宽高比的resize避免变形h,wimage.shape[:2]scalemin(self.input_size[0]/h,self.input_size[1]/w)new_h,new_wint(h*scale),int(w*scale)resizedcv2.resize(image,(new_w,new_h))# 填充到目标尺寸top(self.input_size[0]-new_h)//2bottomself.input_size[0]-new_h-top left(self.input_size[1]-new_w)//2rightself.input_size[1]-new_w-left paddedcv2.copyMakeBorder(resized,top,bottom,left,right,cv2.BORDER_CONSTANT,value(114,114,114)# YOLO常用的灰色填充)# 转BGR如果输入是RGB归一化转CHWiflen(padded.shape)3:# 假设输入是RGB转BGRpaddedcv2.cvtColor(padded,cv2.COLOR_RGB2BGR)blobcv2.dnn.blobFromImage(padded,1/255.0,# 归一化到0-1swapRBFalse,# 已经是BGR了不用再交换cropFalse)returnblob,(scale,(left,top))# 返回缩放比例和填充偏移defpostprocess(self,outputs:np.ndarray,scale:float,padding:Tuple[int,int])-List[dict]: 后处理过滤低置信度框NMS坐标转换回原图 outputs: 模型原始输出 detections[]left_pad,top_padpadding# YOLOv11的输出格式可能是[batch, num_boxes, 85]# 85 [x, y, w, h, conf, class1, class2, ...]fordetinoutputs[0]:scoresdet[5:]class_idnp.argmax(scores)confidencescores[class_id]ifconfidenceself.conf_threshold:# 框坐标是中心点宽高且是相对于输入尺寸的cx,cy,w,hdet[:4]# 转换到填充后的图像坐标x1(cx-w/2-left_pad)/scale y1(cy-h/2-top_pad)/scale x2(cxw/2-left_pad)/scale y2(cyh/2-top_pad)/scale# 确保坐标在图像范围内x1,y1max(0,x1),max(0,y1)x2,y2min(x2,self.input_size[1]/scale),min(y2,self.input_size[0]/scale)detections.append({bbox:[float(x1),float(y1),float(x2),float(y2)],confidence:float(confidence),class_id:int(class_id)})# 简单的NMS - 实际项目建议用torchvision的nmsifdetections:boxesnp.array([d[bbox]fordindetections])scoresnp.array([d[confidence]fordindetections])indicescv2.dnn.NMSBoxes(boxes.tolist(),scores.tolist(),self.conf_threshold,0.45# NMS阈值)iflen(indices)0:detections[detections[i]foriinindices.flatten()]returndetectionsdefinfer(self,image:np.ndarray)-List[dict]: 完整推理流程 返回检测结果列表 # 预处理blob,(scale,padding)self.preprocess(image)# 设置输入self.net.setInput(blob)# 前向推理 - 这里可以加个计时starttime.perf_counter()outputsself.net.forward()infer_time(time.perf_counter()-start)*1000# 转毫秒# 后处理detectionsself.postprocess(outputs,scale,padding)# 可以打印一下耗时print(f推理耗时:{infer_time:.2f}ms, 检测到{len(detections)}个目标)returndetections这个引擎类封装了完整的流程。实际使用中发现预处理和后处理的时间经常比模型推理本身还长所以这两个部分的优化很重要。三、FastAPI服务封装现在用FastAPI把推理引擎包成HTTP服务。# app.pyfromfastapiimportFastAPI,File,UploadFile,HTTPExceptionfromfastapi.responsesimportJSONResponseimportnumpyasnpimportcv2importiofrominference_engineimportYOLOv11EnginefromtypingimportListimportuvicorn appFastAPI(titleYOLOv11推理服务,description基于OpenCV和FastAPI的目标检测API,version1.0)# 全局引擎实例 - 注意线程安全engineNoneapp.on_event(startup)asyncdefstartup_event(): 服务启动时加载模型 这里用全局变量简单处理生产环境要考虑多进程 globalenginetry:engineYOLOv11Engine(yolov11s.onnx)print(模型加载成功服务准备就绪)exceptExceptionase:print(f模型加载失败:{e})raiseapp.post(/predict,response_modelList[dict])asyncdefpredict(file:UploadFileFile(...)): 预测接口 支持jpg、png格式 ifengineisNone:raiseHTTPException(status_code500,detail推理引擎未初始化)# 检查文件类型iffile.content_typenotin[image/jpeg,image/png]:raiseHTTPException(status_code400,detail只支持JPEG和PNG格式)try:# 读取图片contentsawaitfile.read()nparrnp.frombuffer(contents,np.uint8)imagecv2.imdecode(nparr,cv2.IMREAD_COLOR)ifimageisNone:raiseHTTPException(status_code400,detail图片解码失败)# 转RGB前端传的一般是RGBimage_rgbcv2.cvtColor(image,cv2.COLOR_BGR2RGB)# 推理resultsengine.infer(image_rgb)# 返回结果returnJSONResponse(contentresults)exceptExceptionase:print(f推理出错:{e})raiseHTTPException(status_code500,detailf推理过程出错:{str(e)})app.get(/health)asyncdefhealth_check(): 健康检查接口 部署到K8s或Docker时很有用 return{status:healthy,engine_ready:engineisnotNone}if__name____main__:# 生产环境用uvicorn命令行启动这里只是测试uvicorn.run(app,host0.0.0.0,# 监听所有地址port8000,workers1# OpenCV的DNN模块多进程可能有问题建议用1)四、客户端调用示例服务起来了写个客户端测试一下。# client.pyimportrequestsimportcv2importjsondeftest_local_image():# 读取本地图片image_pathtest.jpgimagecv2.imread(image_path)# 转RGBimage_rgbcv2.cvtColor(image,cv2.COLOR_BGR2RGB)# 编码_,img_encodedcv2.imencode(.jpg,image_rgb)# 发送请求urlhttp://localhost:8000/predictfiles{file:(test.jpg,img_encoded.tobytes(),image/jpeg)}responserequests.post(url,filesfiles)ifresponse.status_code200:resultsresponse.json()print(f检测到{len(results)}个目标)# 在图上画框fordetinresults:x1,y1,x2,y2map(int,det[bbox])confdet[confidence]class_iddet[class_id]cv2.rectangle(image,(x1,y1),(x2,y2),(0,255,0),2)labelfClass{class_id}:{conf:.2f}cv2.putText(image,label,(x1,y1-10),cv2.FONT_HERSHEY_SIMPLEX,0.5,(0,255,0),2)cv2.imwrite(result.jpg,image)print(结果已保存到 result.jpg)else:print(f请求失败:{response.status_code})print(response.text)if__name____main__:test_local_image()五、性能优化与生产建议实际部署时遇到几个典型问题内存泄漏OpenCV的DNN模块在某些版本有内存泄漏特别是反复加载不同模型时。解决方案是保持引擎单例不要每次请求都新建。并发问题OpenCV的DNN默认不是线程安全的。如果要用多worker要么用进程隔离每个进程加载一个模型实例要么加锁。实测发现用FastAPI的单个worker配合异步处理吞吐量反而更高。预处理优化图片解码和resize很耗CPU。如果输入图片尺寸固定可以考虑在前端或网关层先做resize。或者用OpenCV的GPU加速如果编译了CUDA支持。批处理支持上面的代码是单张推理。如果请求量大可以攒几个请求一起推理。但OpenCV的DNN对动态batch支持一般需要测试。监控在生产环境一定要加监控。我通常会在/predict接口里埋点记录推理耗时、输入尺寸、检测数量然后推到Prometheus。最后给个经验建议不要过早优化。先让服务跑起来用真实流量压测找到瓶颈再针对性优化。我见过有人花两周优化模型推理时间最后发现80%的时间花在了网络传输和图片解码上。模型部署是个系统工程代码能跑只是第一步。稳定、高效、可维护的服务需要不断调优和打磨。下次我们聊聊用TensorRT做GPU加速部署那又是另一个故事了。