别再硬改源码了！用Flask给YOLOv8加个API，轻松把检测结果推给任何设备
基于Flask构建YOLOv8检测结果转发API的工程实践
在计算机视觉项目的实际落地过程中，我们常常需要将检测结果传递给其他硬件设备。传统做法是直接修改YOLOv8的源代码，但这种硬编码方式存在诸多弊端：代码难以维护、跨平台部署困难、通信方式单一。本文将介绍一种更优雅的解决方案——使用Flask构建轻量级API服务层，实现检测结果与各类硬件设备的灵活对接。
1. 为什么需要解耦YOLOv8与硬件通信？
直接修改YOLOv8核心代码（如plotting.py）来实现串口通信，看似简单实则埋下了不少隐患：
- 代码污染：在视觉算法代码中混入硬件通信逻辑，破坏了代码的单一职责原则
- 维护困难：每次YOLOv8版本更新都需要重新修改和测试通信代码
- 扩展性差：通信方式被固定为串口，难以适应网络、MQTT等其他协议需求
- 部署复杂：需要所有运行环境都具备相同的串口配置
相比之下，API中间层方案具有明显优势：
- 架构清晰：YOLOv8只负责检测，通信由独立服务处理
- 协议灵活：可同时支持串口、Socket、HTTP、MQTT等多种通信方式
- 部署简便：API服务可独立部署在任何设备上
- 调试方便：可通过Postman等工具直接测试接口
2. 系统架构设计
我们的解决方案采用微服务架构，将系统分为三个独立模块：
YOLOv8检测模块 → Flask API服务 → 硬件设备(单片机/树莓派等)2.1 组件交互流程
- YOLOv8完成目标检测后，将结果以HTTP POST请求发送到本地Flask服务
- Flask服务接收JSON格式的检测结果
- 根据配置选择合适的通信方式转发数据
- 硬件设备接收并处理数据
2.2 通信方式对比
| 通信方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 串口通信 | 短距离直连设备 | 简单可靠、低延迟 | 距离短、速率低 |
| Socket | 局域网内设备 | 灵活、支持多客户端 | 需要网络配置 |
| MQTT | 物联网设备 | 支持发布订阅模式 | 需要broker服务器 |
| HTTP | 通用Web服务 | 标准化、跨平台 | 开销较大 |
3. Flask API服务实现
下面我们实现核心的API转发服务，代码采用Python 3.8+和Flask 2.0+。
3.1 基础服务搭建
首先安装所需依赖：
pip install flask flask-cors pyserial paho-mqtt创建主服务文件app.py：
from flask import Flask, request, jsonify import serial import socket import paho.mqtt.client as mqtt app = Flask(__name__) # 通信方式配置 COMM_METHOD = 'serial' # 可选: serial/socket/mqtt/http # 初始化串口 ser = serial.Serial('COM6', 9600, timeout=1) if COMM_METHOD == 'serial' else None @app.route('/api/detections', methods=['POST']) def handle_detections(): data = request.json if not data or 'objects' not in data: return jsonify({'error': 'Invalid data format'}), 400 # 转发检测结果 forward_detections(data['objects']) return jsonify({'status': 'success'}) def forward_detections(objects): """根据配置转发检测结果""" if COMM_METHOD == 'serial': for obj in objects: ser.write(f"{obj['label']},{obj['x']},{obj['y']}\n".encode()) elif COMM_METHOD == 'socket': # Socket转发实现 pass # 其他通信方式实现... if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)3.2 YOLOv8集成改造
修改YOLOv8的检测代码，将结果发送到API服务而非直接操作硬件：
import requests from ultralytics import YOLO model = YOLO('yolov8n.pt') def send_to_api(detections): """将检测结果发送到Flask API""" objects = [{ 'label': det.names[int(cls)], 'conf': float(conf), 'x': float(xyxy[0]), 'y': float(xyxy[1]) } for *xyxy, conf, cls in detections] try: requests.post('http://localhost:5000/api/detections', json={'objects': objects}, timeout=0.5) except requests.exceptions.RequestException: pass # 网络错误处理 results = model(source=0, show=True, stream=True) # 摄像头输入 for result in results: send_to_api(result.boxes)4. 多通信方式实现细节
4.1 串口通信增强实现
基础串口通信存在数据丢失风险，我们需要增加以下改进：
- 数据校验：添加CRC校验位
- 错误重试：失败后自动重发
- 队列缓冲：避免数据拥堵
改进后的串口处理代码：
import crc8 from queue import Queue from threading import Thread serial_queue = Queue(maxsize=100) def serial_worker(): while True: data = serial_queue.get() crc = crc8.crc8() crc.update(data.encode()) packet = f"{data}|{crc.hexdigest()}\n" for _ in range(3): # 最多重试3次 try: ser.write(packet.encode()) break except serial.SerialException: time.sleep(0.1) serial_queue.task_done() Thread(target=serial_worker, daemon=True).start() def forward_detections(objects): for obj in objects: data = f"{obj['label']},{obj['x']},{obj['y']}" serial_queue.put(data)4.2 Socket通信实现
对于需要网络通信的场景，我们可以实现TCP Socket转发：
import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(('192.168.1.100', 8080)) # 目标设备IP和端口 def forward_via_socket(objects): for obj in objects: data = f"{obj['label']},{obj['x']},{obj['y']}\n".encode() sock.sendall(data)4.3 MQTT物联网集成
对于物联网场景，MQTT是更好的选择：
mqtt_client = mqtt.Client() mqtt_client.connect("mqtt.broker.com", 1883) def forward_via_mqtt(objects): for obj in objects: payload = json.dumps(obj) mqtt_client.publish("yolo/detections", payload)5. 性能优化与生产部署
5.1 性能瓶颈分析
在实测中，我们发现主要性能瓶颈来自：
- HTTP API请求的序列化开销
- 检测结果的高频发送导致的网络拥堵
- 串口通信的低速特性
5.2 优化策略
批量发送：将多个检测结果打包发送
# 修改send_to_api函数 def send_to_api(detections): objects = [...] if len(objects) > 0: requests.post('http://localhost:5000/api/detections', json={'objects': objects}, timeout=0.5)异步处理：使用多线程避免阻塞检测流程
from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(max_workers=2) def async_send(objects): try: requests.post('http://localhost:5000/api/detections', json={'objects': objects}, timeout=0.5) except: pass executor.submit(async_send, objects)5.3 生产环境部署建议
使用Gunicorn：提升Flask服务并发能力
gunicorn -w 4 -b :5000 app:app配置Nginx反向代理：提供负载均衡和HTTPS支持
实现服务监控：使用Supervisor管理进程
日志记录：记录所有转发操作以便调试
6. 实际应用案例
6.1 智能垃圾分类系统
通过YOLOv8识别垃圾类型后，API服务将分类指令发送给单片机控制机械臂：
- 可回收物 → 蓝色垃圾桶
- 有害垃圾 → 红色垃圾桶
- 厨余垃圾 → 绿色垃圾桶
- 其他垃圾 → 黑色垃圾桶
6.2 工业质检流水线
检测到产品缺陷时，API服务通过多种方式同时通知：
- 串口信号触发报警灯
- MQTT消息通知中控系统
- Socket消息记录缺陷位置
6.3 智能停车场系统
车牌识别结果通过API服务转发：
- 本地显示屏显示识别结果
- 云端数据库记录车辆进出
- 微信推送停车费通知