RetinafaceCurricularFace实战教程HTTP服务封装思路与Flask轻量API示例1. 为什么需要HTTP服务封装在实际项目中我们很少直接在命令行中运行Python脚本进行人脸识别。更多时候我们需要将模型能力封装成服务供其他系统调用。HTTP API是最常见的集成方式它可以让你的模型被任何编程语言调用Python、Java、JavaScript等轻松部署到服务器提供7x24小时服务支持多用户并发访问方便进行负载均衡和扩展今天我就带你用Flask这个轻量级框架把RetinafaceCurricularFace模型包装成实用的HTTP服务。2. 环境准备与项目结构2.1 确保环境就绪首先进入工作目录并激活环境cd /root/Retinaface_CurricularFace conda activate torch252.2 安装Flask框架pip install flask flask-cors2.3 创建项目结构建议按以下方式组织代码/root/Retinaface_CurricularFace/ ├── app.py # Flask主应用 ├── inference_face.py # 原始推理脚本 ├── api_utils.py # 工具函数 ├── requirements.txt # 依赖列表 └── test_api.py # API测试脚本3. Flask API核心实现3.1 基础API框架创建app.py文件实现最简单的健康检查接口from flask import Flask, request, jsonify from flask_cors import CORS import os import sys # 添加当前目录到Python路径 sys.path.append(os.path.dirname(os.path.abspath(__file__))) app Flask(__name__) CORS(app) # 允许跨域访问 app.route(/health, methods[GET]) def health_check(): 健康检查接口 return jsonify({ status: healthy, message: RetinafaceCurricularFace API is running }) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)3.2 人脸比对API实现创建api_utils.py工具文件import cv2 import numpy as np from PIL import Image import requests from io import BytesIO def load_image(image_input): 加载图片支持本地路径和URL if isinstance(image_input, str): if image_input.startswith((http://, https://)): # 从URL加载图片 response requests.get(image_input) image Image.open(BytesIO(response.content)) else: # 从本地文件加载 image Image.open(image_input) else: # 已经是PIL Image对象 image image_input return cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR) def validate_image(image): 验证图片是否有效 if image is None or image.size 0: return False return True在app.py中添加人脸比对接口from api_utils import load_image, validate_image import subprocess import tempfile import os app.route(/face/compare, methods[POST]) def face_compare(): 人脸比对接口 try: # 获取请求参数 data request.json image1_url data.get(image1) image2_url data.get(image2) threshold data.get(threshold, 0.4) if not image1_url or not image2_url: return jsonify({error: 缺少image1或image2参数}), 400 # 创建临时文件保存图片 with tempfile.NamedTemporaryFile(suffix.jpg, deleteFalse) as f1, \ tempfile.NamedTemporaryFile(suffix.jpg, deleteFalse) as f2: temp_file1 f1.name temp_file2 f2.name # 下载图片到临时文件 img1 load_image(image1_url) img2 load_image(image2_url) if not validate_image(img1) or not validate_image(img2): return jsonify({error: 图片加载失败}), 400 cv2.imwrite(temp_file1, img1) cv2.imwrite(temp_file2, img2) # 调用推理脚本 cmd [ python, inference_face.py, --input1, temp_file1, --input2, temp_file2, --threshold, str(threshold) ] result subprocess.run(cmd, capture_outputTrue, textTrue) # 清理临时文件 os.unlink(temp_file1) os.unlink(temp_file2) if result.returncode ! 0: return jsonify({error: f推理失败: {result.stderr}}), 500 # 解析输出结果 output result.stdout similarity None is_same None for line in output.split(\n): if Similarity Score in line: similarity float(line.split(:)[-1].strip()) elif The two images are of in line: is_same the same person in line if similarity is not None and is_same is not None: return jsonify({ similarity: similarity, is_same_person: is_same, threshold: threshold }) else: return jsonify({error: 结果解析失败}), 500 except Exception as e: return jsonify({error: f服务器内部错误: {str(e)}}), 5003.3 文件上传接口对于本地文件上传添加以下接口from werkzeug.utils import secure_filename import uuid app.route(/face/upload, methods[POST]) def upload_images(): 图片上传接口 if image1 not in request.files or image2 not in request.files: return jsonify({error: 请上传两张图片}), 400 image1 request.files[image1] image2 request.files[image2] # 生成唯一文件名 filename1 f{uuid.uuid4()}_{secure_filename(image1.filename)} filename2 f{uuid.uuid4()}_{secure_filename(image2.filename)} temp_dir /tmp/face_compare os.makedirs(temp_dir, exist_okTrue) filepath1 os.path.join(temp_dir, filename1) filepath2 os.path.join(temp_dir, filename2) image1.save(filepath1) image2.save(filepath2) # 调用比对接口 try: result compare_faces(filepath1, filepath2, float(request.form.get(threshold, 0.4))) # 清理临时文件 os.unlink(filepath1) os.unlink(filepath2) return jsonify(result) except Exception as e: # 确保清理文件 if os.path.exists(filepath1): os.unlink(filepath1) if os.path.exists(filepath2): os.unlink(filepath2) return jsonify({error: str(e)}), 5004. 完整的Flask应用整合所有功能后的完整app.pyfrom flask import Flask, request, jsonify from flask_cors import CORS import subprocess import tempfile import os import sys from api_utils import load_image, validate_image app Flask(__name__) CORS(app) app.route(/health, methods[GET]) def health_check(): return jsonify({status: healthy}) app.route(/face/compare, methods[POST]) def face_compare(): # 实现同上 pass app.route(/face/upload, methods[POST]) def upload_images(): # 实现同上 pass def compare_faces(image1_path, image2_path, threshold0.4): 内部比对函数 cmd [ python, inference_face.py, --input1, image1_path, --input2, image2_path, --threshold, str(threshold) ] result subprocess.run(cmd, capture_outputTrue, textTrue) if result.returncode ! 0: raise Exception(f推理失败: {result.stderr}) output result.stdout similarity None is_same None for line in output.split(\n): if Similarity Score in line: similarity float(line.split(:)[-1].strip()) elif The two images are of in line: is_same the same person in line if similarity is not None and is_same is not None: return { similarity: similarity, is_same_person: is_same, threshold: threshold } else: raise Exception(结果解析失败) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)5. 测试你的API服务5.1 启动服务python app.py服务将在 http://0.0.0.0:5000 启动5.2 使用curl测试创建测试脚本test_api.pyimport requests import json # 测试健康检查 response requests.get(http://localhost:5000/health) print(健康检查:, response.json()) # 测试URL图片比对 payload { image1: https://example.com/path/to/image1.jpg, image2: https://example.com/path/to/image2.jpg, threshold: 0.4 } response requests.post(http://localhost:5000/face/compare, jsonpayload) print(URL比对结果:, response.json())5.3 使用Python客户端测试import requests def test_face_comparison(image1_path, image2_path, threshold0.4): 测试人脸比对API with open(image1_path, rb) as f1, open(image2_path, rb) as f2: files { image1: f1, image2: f2 } data {threshold: str(threshold)} response requests.post(http://localhost:5000/face/upload, filesfiles, datadata) return response.json() # 使用示例 result test_face_comparison(path/to/photo1.jpg, path/to/photo2.jpg) print(result)6. 生产环境部署建议6.1 使用Gunicorn部署安装Gunicornpip install gunicorn启动服务gunicorn -w 4 -b 0.0.0.0:5000 app:app6.2 添加错误处理和日志增强API的错误处理import logging from logging.handlers import RotatingFileHandler # 配置日志 if not app.debug: handler RotatingFileHandler(app.log, maxBytes10000, backupCount3) handler.setLevel(logging.INFO) app.logger.addHandler(handler) app.errorhandler(404) def not_found(error): return jsonify({error: 接口不存在}), 404 app.errorhandler(500) def internal_error(error): app.logger.error(f服务器错误: {error}) return jsonify({error: 内部服务器错误}), 5006.3 添加速率限制防止API被滥用pip install flask-limiterfrom flask_limiter import Limiter from flask_limiter.util import get_remote_address limiter Limiter( appapp, key_funcget_remote_address, default_limits[200 per day, 50 per hour] ) app.route(/face/compare, methods[POST]) limiter.limit(10 per minute) def face_compare(): # 接口实现 pass7. 总结通过这个教程你已经学会了如何将RetinafaceCurricularFace模型封装成HTTP服务。关键要点包括Flask基础框架搭建轻量级Web服务API设计设计合理的接口参数和返回格式错误处理完善的异常捕获和日志记录文件处理支持URL和文件上传两种方式生产优化使用Gunicorn部署和添加速率限制现在你的模型已经不再是孤立的脚本而是一个可以集成到任何系统中的服务了。无论是Web应用、移动App还是其他后端系统都可以通过简单的HTTP调用来使用强大的人脸识别能力。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。