M2FP API调用示例：Python requests实现批量人像解析自动化

M2FP API调用示例：Python requests实现批量人像解析自动化

📖 项目简介：M2FP 多人人体解析服务

在计算机视觉领域，人体语义分割（Human Parsing）是一项关键的细粒度图像理解任务，广泛应用于虚拟试衣、智能安防、AR/VR内容生成等场景。传统的语义分割模型往往难以处理多人、遮挡、姿态复杂的情况，而M2FP (Mask2Former-Parsing)模型正是为此类挑战而生。

M2FP 基于 ModelScope 平台构建，采用先进的Mask2Former 架构，结合 ResNet-101 骨干网络，在多人人体解析任务上表现出卓越的精度与鲁棒性。它不仅能识别图像中的多个个体，还能将每个人的身体划分为多达 20 个语义区域，如面部、头发、左臂、右腿、上衣、裤子等，并输出像素级的掩码（mask）结果。

更进一步，该服务封装为一个完整的WebUI + RESTful API 系统镜像，内置 Flask 框架提供可视化交互界面和标准化接口访问能力。特别值得一提的是其自动拼图算法——模型原始输出为一系列二值掩码，系统会自动将其按预设颜色表叠加融合，生成一张直观的彩色语义分割图，极大提升了可读性和实用性。

💡 核心亮点总结： - ✅高精度多人解析：支持多目标检测与精细化部位分割 - ✅开箱即用 WebUI：无需编码即可上传图片查看结果 - ✅稳定运行环境：锁定 PyTorch 1.13.1 + MMCV-Full 1.7.1，彻底规避常见兼容性问题 - ✅CPU 友好设计：专为无 GPU 环境优化，推理流畅不卡顿 - ✅API 接口开放：支持通过 HTTP 请求进行自动化调用，便于集成到生产流程

🧩 技术架构与工作逻辑拆解

要实现“批量人像解析自动化”，我们不能仅依赖 WebUI 手动操作。真正的工程价值在于利用其暴露的REST API 接口，通过脚本化方式完成大规模图像处理任务。

整个系统的运作流程如下：

1. 前端请求发起：用户通过 WebUI 或外部程序发送POST /predict请求，携带待解析图像。
2. 后端接收处理：Flask 服务接收到图像数据后，调用 M2FP 模型执行推理。
3. 模型输出掩码：模型返回每个身体部位的二值掩码列表（JSON 格式），包含类别标签与 base64 编码的 mask 图像。
4. 拼图算法合成：服务端内置算法将所有 mask 按颜色映射表合并成一张彩色分割图。
5. 响应返回结果：最终返回原始 mask 列表 + 合成后的可视化图像（base64 编码或 URL）。

这种设计使得我们可以绕过图形界面，直接使用 Python 的requests库模拟 HTTP 请求，实现全自动化的批量解析。

🛠️ 实践应用：基于 Python requests 的批量调用方案

1. 技术选型理由

为什么选择requests？因为它具备以下优势：

| 特性 | 说明 | |------|------| |简洁易用| 接口清晰，几行代码即可完成 POST 请求 | |广泛兼容| 支持文件上传、JSON 数据传输、base64 编码等常用格式 | |易于批量化| 可结合os.listdir()、concurrent.futures实现并发处理 | |调试友好| 易于打印响应内容、状态码、耗时等信息 |

相比 Selenium 自动化点击 WebUI，requests更轻量、高效且适合部署在服务器端长期运行。

2. 调用前准备：获取 API 地址与测试连通性

假设你已成功启动 M2FP 镜像服务，通常可通过平台提供的 HTTP 入口访问 WebUI，例如地址为：

http://192.168.1.100:8080

我们需要确认两个关键接口是否存在：

• GET /→ WebUI 主页（用于验证服务是否正常）
• POST /predict→ 图像解析接口（核心功能）

先编写一段简单代码测试连接：

import requests # 替换为你的实际服务地址 BASE_URL = "http://192.168.1.100:8080" try: response = requests.get(f"{BASE_URL}/", timeout=10) if response.status_code == 200: print("✅ 服务连接成功！") else: print(f"❌ 服务返回异常状态码：{response.status_code}") except Exception as e: print(f"❌ 连接失败：{e}")

若输出 “✅ 服务连接成功！”，说明服务正常运行，可以继续下一步。

3. 核心实现：调用/predict接口完成单次解析

以下是调用 API 解析一张图片的核心代码：

import requests import json import base64 from PIL import Image from io import BytesIO import os def call_m2fp_api(image_path, api_url): """ 调用 M2FP API 完成人像解析 :param image_path: 本地图片路径 :param api_url: API 地址，如 http://xxx/predict :return: 返回解析结果（dict） """ # 读取图像并编码为 base64 with open(image_path, 'rb') as f: img_data = f.read() img_base64 = base64.b64encode(img_data).decode('utf-8') # 构造请求体 payload = { "image": img_base64 } headers = { "Content-Type": "application/json" } try: response = requests.post(api_url, data=json.dumps(payload), headers=headers, timeout=30) if response.status_code == 200: result = response.json() return result else: print(f"[错误] HTTP {response.status_code}: {response.text}") return None except Exception as e: print(f"[异常] 请求失败: {e}") return None # 使用示例 API_URL = f"{BASE_URL}/predict" result = call_m2fp_api("test_person.jpg", API_URL) if result: print("🟢 解析成功！共检测到 {} 个人体实例".format(len(result['masks']))) # 输出部分字段示例 for i, mask_info in enumerate(result['masks']): print(f" 实例 {i+1}: 类别={mask_info['label']}, 置信度={mask_info['score']:.3f}") else: print("🔴 解析失败，请检查输入或服务状态")

🔍 代码解析：

• base64 编码：将二进制图像转为字符串，便于 JSON 传输
• Content-Type 设置：必须设置为application/json，否则 Flask 可能无法正确解析
• 超时控制：CPU 推理较慢，建议设置timeout=30防止阻塞
• 错误捕获：涵盖网络异常、服务崩溃、格式错误等多种情况

4. 批量处理：遍历目录实现自动化流水线

接下来我们将上述函数扩展为批量处理脚本，支持对整个文件夹内的图片进行解析，并保存结果。

import os from concurrent.futures import ThreadPoolExecutor import time def save_result_images(result, output_dir, filename_prefix): """ 保存可视化分割图和原始 mask """ if not os.path.exists(output_dir): os.makedirs(output_dir) # 保存合成后的可视化图像 if 'colored_mask' in result and result['colored_mask']: colored_img_data = base64.b64decode(result['colored_mask']) colored_img = Image.open(BytesIO(colored_img_data)) colored_img.save(os.path.join(output_dir, f"{filename_prefix}_vis.png")) # 保存每个原始 mask（可选） masks_dir = os.path.join(output_dir, "masks") os.makedirs(masks_dir, exist_ok=True) for idx, m in enumerate(result['masks']): mask_data = base64.b64decode(m['mask']) mask_img = Image.open(BytesIO(mask_data)) mask_img.save(os.path.join(masks_dir, f"{filename_prefix}_mask_{idx}_{m['label']}.png")) def process_single_image(image_path, api_url, output_dir): """ 处理单张图像并保存结果 """ print(f"🔄 正在处理: {image_path}") start_time = time.time() result = call_m2fp_api(image_path, api_url) if result: filename = os.path.basename(image_path) name_only = os.path.splitext(filename)[0] save_result_images(result, output_dir, name_only) cost = time.time() - start_time print(f"✅ 成功处理 {filename}，耗时 {cost:.2f}s") return True else: print(f"❌ 处理失败: {image_path}") return False def batch_process(input_folder, output_folder, api_url, max_workers=3): """ 批量处理指定目录下所有图像 """ image_exts = ('.jpg', '.jpeg', '.png', '.bmp') image_files = [ os.path.join(input_folder, f) for f in os.listdir(input_folder) if f.lower().endswith(image_exts) ] if not image_files: print("⚠️ 未找到任何图像文件") return print(f"📦 共发现 {len(image_files)} 张图像，开始批量解析...") with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [ executor.submit(process_single_image, img_file, api_url, output_folder) for img_file in image_files ] success_count = sum(future.result() for future in futures) print(f"🎉 批量处理完成！成功解析 {success_count}/{len(image_files)} 张图像") # 调用批量处理 batch_process( input_folder="./input_images", output_folder="./output_results", api_url=f"{BASE_URL}/predict", max_workers=3 # 根据 CPU 性能调整并发数 )

5. 实践难点与优化建议

在真实项目中，我们遇到了以下几个典型问题及解决方案：

❌ 问题1：CPU 推理速度慢导致超时

• 现象：单张图像推理超过 20 秒，requests默认超时为 10 秒，导致中断
• 解决：显式设置timeout=60，并降低并发数（max_workers=2~3）

❌ 问题2：内存占用过高引发 OOM

• 现象：连续处理百张图像时内存持续增长
• 解决：
• 使用del result及时释放变量
• 添加gc.collect()强制垃圾回收
• 分批次处理（每 10 张 sleep 1s）

⚙️ 优化建议

| 优化项 | 建议 | |-------|------| |并发控制| CPU 环境建议max_workers ≤ 4，避免资源争抢 | |日志记录| 将成功/失败记录写入 CSV 文件，便于追踪 | |断点续传| 记录已完成文件名，防止重复处理 | |结果缓存| 对相同哈希值的图像跳过重复计算 |

📊 应用场景与落地价值

该自动化方案已在多个实际项目中落地：

• 电商模特图分析：自动提取服装区域，用于商品属性标注
• 健身动作评估系统：结合姿态估计，分析肢体运动轨迹
• 数字人内容生成：为虚拟形象换装提供精准蒙版支持
• 安防行为识别：辅助判断人员着装、携带物品等特征

通过 API 批量调用，原本需要人工逐张上传的操作被压缩至几分钟内完成，效率提升数十倍。

✅ 最佳实践总结

1. 优先使用 API 而非 WebUI：自动化场景下 API 是唯一可行路径
2. 合理设置超时与重试机制：增加retry_on_failure逻辑提升稳定性
3. 监控资源使用：定期检查 CPU、内存、磁盘空间
4. 结构化存储结果：将 JSON 结果与图像分类归档，便于后续分析
5. 封装为微服务模块：可作为独立人体解析服务嵌入更大系统

🎯 下一步学习建议

• 学习如何使用gunicorn + nginx部署 Flask API 提升吞吐量
• 探索将 M2FP 模型导出为 ONNX 格式以进一步加速推理
• 结合 OpenPose 实现“人体解析 + 关键点”联合分析 pipeline

📌 核心结论：
M2FP 不仅是一个高精度的人体解析模型，更是一套完整的工程化解决方案。通过 Pythonrequests调用其 API，我们能够轻松实现从“手动测试”到“工业级批量处理”的跨越，真正发挥 AI 模型的生产力价值。